Skip to content
Open
29 changes: 29 additions & 0 deletions workflow/test_worker.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
import unittest
Copy link
Copy Markdown

@cubic-dev-ai cubic-dev-ai bot Mar 31, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: This test file is outside the configured pytest testpaths ("tests" and "ai/core/tests"), so it won’t be collected or run in CI. Move it under the tests/ hierarchy or update pytest testpaths to include workflow/ so the coverage actually runs.

Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At workflow/test_worker.py, line 1:

<comment>This test file is outside the configured pytest testpaths ("tests" and "ai/core/tests"), so it won’t be collected or run in CI. Move it under the tests/ hierarchy or update pytest testpaths to include workflow/ so the coverage actually runs.</comment>

<file context>
@@ -0,0 +1,29 @@
+import unittest
+import os
+import tempfile
</file context>
Fix with Cubic

import os
import tempfile
import json
import worker
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

workflow/test_worker.py importing worker directly is fragile: depending on how tests are executed (repo root vs workflow/ as CWD), this can import the wrong module or fail. It’s safer to import via the package/module path used by the application (e.g., from workflow import worker or a relative import).

Suggestion

Prefer an absolute/relative package import so the test runs reliably under CI runners invoked from the repo root.

from workflow import worker
# or, if `workflow` is a package:
from . import worker

Reply with "@CharlieHelps yes please" if you’d like me to add a commit with this change.


class TestWorker(unittest.TestCase):
def setUp(self):
self.fd, self.temp_path = tempfile.mkstemp()
self.original_log_path = worker.LOG_PATH
worker.LOG_PATH = self.temp_path

Comment on lines +9 to +12
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

tempfile.mkstemp() returns an open file descriptor. Keeping it open while worker.log_event() opens/writes the same path can fail on Windows due to file locking/sharing semantics. Close the FD immediately after creating the path (or use NamedTemporaryFile(delete=False) and close it).

Suggestion

Close the mkstemp() descriptor right away (and set fd=None so tearDown doesn’t double-close), or switch to NamedTemporaryFile.

def setUp(self):
    self.fd, self.temp_path = tempfile.mkstemp()
    os.close(self.fd)
    self.fd = None
    self.original_log_path = worker.LOG_PATH
    worker.LOG_PATH = self.temp_path

def tearDown(self):
    worker.LOG_PATH = self.original_log_path
    if self.fd is not None:
        os.close(self.fd)
    os.remove(self.temp_path)

Reply with "@CharlieHelps yes please" if you’d like me to add a commit with this suggestion.

def tearDown(self):
worker.LOG_PATH = self.original_log_path
os.close(self.fd)
os.remove(self.temp_path)

def test_log_event(self):
worker.log_event("evt_123", "test message")
with open(self.temp_path, "r") as f:
lines = f.readlines()
self.assertEqual(len(lines), 1)
data = json.loads(lines[0])
self.assertEqual(data["event_id"], "evt_123")
self.assertEqual(data["message"], "test message")
self.assertIn("timestamp", data)

if __name__ == "__main__":
unittest.main()
Loading