Fixes to fifo

python_files/vscode_pytest/__init__.py

@@ -20,17 +20,6 @@
 
 import pytest
 
-
-# sys.path.append("/Users/eleanorboyd/vscode-python/.nox/install_python_libs/lib/python3.10")
-# sys.path.append("/Users/eleanorboyd/vscode-python-debugger")
-# sys.path.append("/Users/eleanorboyd/vscode-python-debugger/bundled")
-# sys.path.append("/Users/eleanorboyd/vscode-python-debugger/bundled/libs")
-
-# import debugpy  # noqa: E402
-
-# debugpy.connect(5678)
-# debugpy.breakpoint()  # noqa: E702
-
 if TYPE_CHECKING:
     from pluggy import Result
 
@@ -161,7 +150,7 @@ def pytest_exception_interact(node, call, report):
             collected_test = TestRunResultDict()
             collected_test[node_id] = item_result
             cwd = pathlib.Path.cwd()
-            execution_post(
+            send_execution_message(
                 os.fsdecode(cwd),
                 "success",
                 collected_test if collected_test else None,
@@ -285,7 +274,7 @@ def pytest_report_teststatus(report, config):  # noqa: ARG001
             )
             collected_test = TestRunResultDict()
             collected_test[absolute_node_id] = item_result
-            execution_post(
+            send_execution_message(
                 os.fsdecode(cwd),
                 "success",
                 collected_test if collected_test else None,
@@ -319,7 +308,7 @@ def pytest_runtest_protocol(item, nextitem):  # noqa: ARG001
             )
             collected_test = TestRunResultDict()
             collected_test[absolute_node_id] = item_result
-            execution_post(
+            send_execution_message(
                 os.fsdecode(cwd),
                 "success",
                 collected_test if collected_test else None,
@@ -390,15 +379,15 @@ def pytest_sessionfinish(session, exitstatus):
                 "children": [],
                 "id_": "",
             }
-            post_response(os.fsdecode(cwd), error_node)
+            send_discovery_message(os.fsdecode(cwd), error_node)
         try:
             session_node: TestNode | None = build_test_tree(session)
             if not session_node:
                 raise VSCodePytestError(
                     "Something went wrong following pytest finish, \
                         no session node was created"
                 )
-            post_response(os.fsdecode(cwd), session_node)
+            send_discovery_message(os.fsdecode(cwd), session_node)
         except Exception as e:
             ERRORS.append(
                 f"Error Occurred, traceback: {(traceback.format_exc() if e.__traceback__ else '')}"
@@ -410,7 +399,7 @@ def pytest_sessionfinish(session, exitstatus):
                 "children": [],
                 "id_": "",
             }
-            post_response(os.fsdecode(cwd), error_node)
+            send_discovery_message(os.fsdecode(cwd), error_node)
     else:
         if exitstatus == 0 or exitstatus == 1:
             exitstatus_bool = "success"
@@ -420,15 +409,15 @@ def pytest_sessionfinish(session, exitstatus):
             )
             exitstatus_bool = "error"
 
-            execution_post(
+            send_execution_message(
                 os.fsdecode(cwd),
                 exitstatus_bool,
                 None,
             )
         # send end of transmission token
     command_type = "discovery" if IS_DISCOVERY else "execution"
     payload: EOTPayloadDict = {"command_type": command_type, "eot": True}
-    send_post_request(payload)
+    send_message(payload)
 
 
 def build_test_tree(session: pytest.Session) -> TestNode:
@@ -790,8 +779,10 @@ def get_node_path(node: Any) -> pathlib.Path:
 atexit.register(lambda: __writer.close() if __writer else None)
 
 
-def execution_post(cwd: str, status: Literal["success", "error"], tests: TestRunResultDict | None):
-    """Sends a POST request with execution payload details.
+def send_execution_message(
+    cwd: str, status: Literal["success", "error"], tests: TestRunResultDict | None
+):
+    """Sends message execution payload details.
 
     Args:
         cwd (str): Current working directory.
@@ -803,10 +794,10 @@ def execution_post(cwd: str, status: Literal["success", "error"], tests: TestRun
     )
     if ERRORS:
         payload["error"] = ERRORS
-    send_post_request(payload)
+    send_message(payload)
 
 
-def post_response(cwd: str, session_node: TestNode) -> None:
+def send_discovery_message(cwd: str, session_node: TestNode) -> None:
     """
     Sends a POST request with test session details in payload.
 
@@ -822,7 +813,7 @@ def post_response(cwd: str, session_node: TestNode) -> None:
     }
     if ERRORS is not None:
         payload["error"] = ERRORS
-    send_post_request(payload, cls_encoder=PathEncoder)
+    send_message(payload, cls_encoder=PathEncoder)
 
 
 class PathEncoder(json.JSONEncoder):
@@ -834,7 +825,7 @@ def default(self, obj):
         return super().default(obj)
 
 
-def send_post_request(
+def send_message(
     payload: ExecutionPayloadDict | DiscoveryPayloadDict | EOTPayloadDict,
     cls_encoder=None,
 ):
@@ -845,7 +836,6 @@ def send_post_request(
     payload -- the payload data to be sent.
     cls_encoder -- a custom encoder if needed.
     """
-    print("EJFB into send post request!")
     if not TEST_RUN_PIPE:
         error_msg = (
             "PYTEST ERROR: TEST_RUN_PIPE is not set at the time of pytest starting. "
@@ -860,10 +850,8 @@ def send_post_request(
 
     if __writer is None:
         try:
-            print("EJFB attemping writer open")
             __writer = open(TEST_RUN_PIPE, "w", encoding="utf-8", newline="\r\n")  # noqa: SIM115, PTH123
         except Exception as error:
-            print("EJFB error in writer open")
             error_msg = f"Error attempting to connect to extension named pipe {TEST_RUN_PIPE}[vscode-pytest]: {error}"
             print(error_msg, file=sys.stderr)
             print(
@@ -880,16 +868,11 @@ def send_post_request(
         "params": payload,
     }
     data = json.dumps(rpc, cls=cls_encoder)
-    print(f"EJFB Plugin info[vscode-pytest]: sending data: \n{data}\n")
     try:
         if __writer:
             request = f"""content-length: {len(data)}\ncontent-type: application/json\n\n{data}"""
             __writer.write(request)
             __writer.flush()
-            print(
-                f"EJFB Plugin info[vscode-pytest]: data sent successfully[vscode-pytest]: \n{data}\n"
-            )
-            # __writer.close()
         else:
             print(
                 f"Plugin error connection error[vscode-pytest], writer is None \n[vscode-pytest] data: \n{data} \n",

src/client/common/pipes/namedPipes.ts

@@ -3,12 +3,12 @@
 
 import * as cp from 'child_process';
 import * as crypto from 'crypto';
-import * as fs from 'fs';
+import * as fs from 'fs-extra';
 import * as net from 'net';
 import * as os from 'os';
 import * as path from 'path';
 import * as rpc from 'vscode-jsonrpc/node';
-import { CancellationError, CancellationToken } from 'vscode';
+import { CancellationError, CancellationToken, Disposable } from 'vscode';
 import { traceVerbose } from '../../logging';
 import { isWindows } from '../platform/platformService';
 import { createDeferred } from '../utils/async';
@@ -73,6 +73,9 @@ export async function createWriterPipe(pipeName: string, token?: CancellationTok
     }
     // linux implementation of FIFO
     await mkfifo(pipeName);
+    try {
+        await fs.chmod(pipeName, 0o666);
+    } catch {}
     const writer = fs.createWriteStream(pipeName, {
         encoding: 'utf-8',
     });
@@ -86,14 +89,14 @@ class CombinedReader implements rpc.MessageReader {
 
     private _onPartialMessage = new rpc.Emitter<rpc.PartialMessageInfo>();
 
-    private _listeners = new rpc.Emitter<rpc.NotificationMessage>();
-
-    private _readers: rpc.MessageReader[] = [];
+    private _callback: rpc.DataCallback = () => {};
 
     private _disposables: rpc.Disposable[] = [];
 
+    private _readers: rpc.MessageReader[] = [];
+
     constructor() {
-        this._disposables.push(this._onClose, this._onError, this._onPartialMessage, this._listeners);
+        this._disposables.push(this._onClose, this._onError, this._onPartialMessage);
     }
 
     onError: rpc.Event<Error> = this._onError.event;
@@ -103,34 +106,41 @@ class CombinedReader implements rpc.MessageReader {
     onPartialMessage: rpc.Event<rpc.PartialMessageInfo> = this._onPartialMessage.event;
 
     listen(callback: rpc.DataCallback): rpc.Disposable {
-        return this._listeners.event(callback);
+        this._callback = callback;
+        return new Disposable(() => (this._callback = () => {}));
     }
 
     add(reader: rpc.MessageReader): void {
         this._readers.push(reader);
-        this._disposables.push(
-            reader.onError((error) => this._onError.fire(error)),
-            reader.onClose(() => this.dispose()),
-            reader.onPartialMessage((info) => this._onPartialMessage.fire(info)),
-            reader.listen((msg) => {
-                this._listeners.fire(msg as rpc.NotificationMessage);
-            }),
-        );
+        reader.listen((msg) => {
+            this._callback(msg as rpc.NotificationMessage);
+        });
+        this._disposables.push(reader);
+        reader.onClose(() => {
+            this.remove(reader);
+            if (this._readers.length === 0) {
+                this._onClose.fire();
+            }
+        });
+        reader.onError((e) => {
+            this.remove(reader);
+            this._onError.fire(e);
+        });
     }
 
-    error(error: Error): void {
-        this._onError.fire(error);
+    remove(reader: rpc.MessageReader): void {
+        const found = this._readers.find((r) => r === reader);
+        if (found) {
+            this._readers = this._readers.filter((r) => r !== reader);
+            reader.dispose();
+        }
     }
 
     dispose(): void {
-        this._onClose.fire();
-        this._disposables.forEach((disposable) => {
-            try {
-                disposable.dispose();
-            } catch (e) {
-                /* noop */
-            }
-        });
+        this._readers.forEach((r) => r.dispose());
+        this._readers = [];
+        this._disposables.forEach((disposable) => disposable.dispose());
+        this._disposables = [];
     }
 }
 
@@ -166,10 +176,11 @@ export async function createReaderPipe(pipeName: string, token?: CancellationTok
         deferred.resolve(combined);
         return deferred.promise;
     }
-    // linux implementation of FIFO
+    // mac/linux implementation of FIFO
     await mkfifo(pipeName);
-    const reader = fs.createReadStream(pipeName, {
-        encoding: 'utf-8',
-    });
+    try {
+        await fs.chmod(pipeName, 0o666);
+    } catch {}
+    const reader = fs.createReadStream(pipeName, { encoding: 'utf-8' });
     return new rpc.StreamMessageReader(reader, 'utf-8');
 }