merge: merge origin/feature/teamhoa/r05-r06 (R07/R08) into feature/delta-team/epic-R04

This commit is contained in:
2026-08-28 10:32:38 +09:00
72 changed files with 6546 additions and 2657 deletions
+142
View File
@@ -0,0 +1,142 @@
"""EPIC R08-T13: TokenUsageCardWidget / UsageChartWidget / HabitsWidget /
DashboardTab shell, real Qt offscreen.
``DashboardQueryService``'s own logic is unit tested (R08-T13, no Qt) in
``tests/unit/test_dashboard_query_service.py``; this file proves the three
widgets and the shell are actually wired to it and to each other (the period
selector living on ``UsageChartWidget`` driving all three).
"""
from __future__ import annotations
import json
import os
from datetime import date
import pytest
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from cowork_local.config import AppConfig # noqa: E402
from cowork_local.core import usage_tracker as ut # noqa: E402
from cowork_local.state import AppContext # noqa: E402
pytest.importorskip("PySide6", reason="Qt is required for the integration suite")
@pytest.fixture(scope="module")
def qt_app():
from PySide6.QtWidgets import QApplication
return QApplication.instance() or QApplication([])
@pytest.fixture
def usage_dir(tmp_path, monkeypatch):
d = tmp_path / "usage"
monkeypatch.setattr(ut, "USAGE_DIR", d)
return d
@pytest.fixture
def ctx(qt_app, tmp_path):
return AppContext(AppConfig.load(tmp_path / "config.json"))
def _write_event(usage_dir, day: date, **overrides):
usage_dir.mkdir(parents=True, exist_ok=True)
event = {
"ts": f"{day.isoformat()}T10:00:00", "source": "cowork", "label": "Test chat",
"provider": "anthropic", "model": "claude-sonnet-4-6",
"in": 100, "out": 50, "cache": 0, "estimated": False,
"account": "", "machine": "",
}
event.update(overrides)
path = usage_dir / f"{day.isoformat()}.jsonl"
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(event) + "\n")
def test_dashboard_tab_builds_and_populates_cards(usage_dir, ctx):
from cowork_local.presentation.dashboard.dashboard_tab import DashboardTab
_write_event(usage_dir, date.today())
tab = DashboardTab(ctx) # refresh() runs once at construction
assert "100" in tab.token_cards.card_in.value_lbl.text() \
or tab.token_cards.card_in.value_lbl.text() # non-crashing, has SOME text
assert tab.token_cards.card_total.value_lbl.text() != ""
def test_period_navigation_on_chart_refreshes_the_whole_shell(usage_dir, ctx):
from cowork_local.presentation.dashboard.dashboard_tab import DashboardTab
_write_event(usage_dir, date.today())
tab = DashboardTab(ctx)
calls = []
tab.refresh = lambda *a, orig=tab.refresh: (calls.append(1), orig(*a))[-1]
tab.chart._chart_prev()
assert calls == [1]
def test_granularity_change_resets_offset_to_current(usage_dir, ctx):
from cowork_local.presentation.dashboard.dashboard_tab import DashboardTab
from cowork_local.presentation.dashboard.usage_chart_widget import UsageChartWidget
tab = DashboardTab(ctx)
tab.chart._chart_offset = -3
idx = tab.chart.gran_combo.findData("month")
tab.chart.gran_combo.setCurrentIndex(idx)
assert tab.chart.chart_offset == 0
def test_currency_change_persists_and_triggers_refresh(usage_dir, ctx):
from cowork_local.presentation.dashboard.dashboard_tab import DashboardTab
tab = DashboardTab(ctx)
idx = tab.chart.currency_combo.findData("EUR") \
if tab.chart.currency_combo.findData("EUR") >= 0 else 0
seen = []
tab.chart.currency_changed.connect(lambda: seen.append(1))
tab.chart.currency_combo.setCurrentIndex(idx)
if tab.chart.currency_combo.currentData() != "USD":
assert seen == [1]
assert ctx.config.data["usage"]["currency"] == tab.chart.currency_combo.currentData()
def test_habits_widget_ai_analyze_noop_without_data(usage_dir, ctx):
"""No events in range -> emits a status message instead of starting a
background worker (matches the original _ai_analyze guard)."""
from cowork_local.presentation.dashboard.habits_widget import HabitsWidget
from cowork_local.application.monitoring import DashboardQueryService
query = DashboardQueryService(ctx)
widget = HabitsWidget(ctx, query)
widget.refresh(date(2020, 1, 1), date(2020, 1, 1))
messages = []
widget.status_message.connect(messages.append)
widget._ai_analyze()
assert messages # "no data" status, no worker started
assert widget._ai_worker is None
def test_budget_apply_updates_budget_card(usage_dir, ctx):
from cowork_local.application.monitoring import DashboardQueryService
from cowork_local.presentation.dashboard.token_usage_card_widget import TokenUsageCardWidget
query = DashboardQueryService(ctx)
widget = TokenUsageCardWidget(ctx, query)
widget.budget_card.budget_spin.setValue(50.0)
widget._apply_budget()
status = query.budget_status()
assert status is not None
assert status["amount_usd"] == pytest.approx(50.0)
+193
View File
@@ -0,0 +1,193 @@
"""EPIC R08-T12: WorkspaceFileTree / DocumentPreviewManager / FolderTab,
real Qt offscreen.
Scope note: the AI-Edit panel's send -> plan -> edit -> apply pipeline
(``ai_edit_pipeline.py``) runs on a real ``core.worker.AgentWorker`` QThread
and had ZERO existing tests before this task (confirmed by grep — nothing
under tests/ exercised ``ui/folder_tab.py``'s AI methods except the routing
surface test, fixed alongside this task in
``tests/integration/test_routing_surfaces.py``). Driving that full async
pipeline end to end is out of scope here; what's covered is everything that
doesn't need a live QThread: navigation, preview rendering, and — the
concrete R08-T12 deliverable — that saves/creates actually go through
``FileWorkspaceService`` (containment enforced, not just "writes a file").
"""
from __future__ import annotations
import os
from pathlib import Path
import pytest
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from cowork_local.config import AppConfig # noqa: E402
from cowork_local.state import AppContext # noqa: E402
pytest.importorskip("PySide6", reason="Qt is required for the integration suite")
@pytest.fixture(scope="module")
def qt_app():
from PySide6.QtWidgets import QApplication
return QApplication.instance() or QApplication([])
@pytest.fixture
def ctx(qt_app, tmp_path: Path):
return AppContext(AppConfig.load(tmp_path / "config.json"))
@pytest.fixture
def root(tmp_path: Path) -> Path:
d = tmp_path / "workspace"
d.mkdir()
return d
# ---- WorkspaceFileTree ----------------------------------------------------- #
def test_workspace_file_tree_set_root_updates_label_and_model(qt_app, root):
from cowork_local.presentation.folder.workspace_file_tree import WorkspaceFileTree
tree = WorkspaceFileTree(str(root))
other = root.parent / "other-root"
other.mkdir()
tree.set_root(str(other))
assert tree.root == str(other)
assert tree.path_lbl.text() == str(other)
def test_workspace_file_tree_ignores_an_invalid_root(qt_app, root):
from cowork_local.presentation.folder.workspace_file_tree import WorkspaceFileTree
tree = WorkspaceFileTree(str(root))
tree.set_root(str(root / "does-not-exist"))
assert tree.root == str(root) # unchanged
def test_workspace_file_tree_click_emits_file_selected(qt_app, root):
from cowork_local.presentation.folder.workspace_file_tree import WorkspaceFileTree
f = root / "a.txt"
f.write_text("hello", encoding="utf-8")
tree = WorkspaceFileTree(str(root))
seen = []
tree.file_selected.connect(seen.append)
index = tree.model.index(str(f))
tree._on_tree_clicked(index)
assert seen and os.path.normpath(seen[0]) == os.path.normpath(str(f))
# ---- DocumentPreviewManager ------------------------------------------------- #
def test_open_file_renders_code_into_the_editor(qt_app, root):
from cowork_local.presentation.folder.document_preview_manager import DocumentPreviewManager
f = root / "script.py"
f.write_text("print('hi')\n", encoding="utf-8")
preview = DocumentPreviewManager(str(root))
preview.open_file(str(f))
assert preview.stack.currentWidget() is preview.editor
assert preview.editor.toPlainText() == "print('hi')\n"
assert preview.current_file == str(f)
def test_open_file_on_a_binary_file_shows_the_placeholder(qt_app, root):
from cowork_local.presentation.folder.document_preview_manager import DocumentPreviewManager
f = root / "data.bin"
f.write_bytes(b"\x00\x01\x02binary")
preview = DocumentPreviewManager(str(root))
preview.open_file(str(f))
assert preview.stack.currentWidget() is preview._placeholder
def test_save_writes_through_file_workspace_service(qt_app, root):
"""The concrete R08-T12/R06-T05 deliverable: FileWorkspaceService (built
at R06, unused until this task) is now the actual write path."""
from cowork_local.presentation.folder.document_preview_manager import DocumentPreviewManager
f = root / "note.txt"
f.write_text("old", encoding="utf-8")
preview = DocumentPreviewManager(str(root))
preview.open_file(str(f))
preview.editor.setPlainText("new content")
preview.save()
assert f.read_text(encoding="utf-8") == "new content"
def test_create_new_file_writes_inside_root_via_file_workspace_service(qt_app, root):
from cowork_local.presentation.folder.document_preview_manager import DocumentPreviewManager
preview = DocumentPreviewManager(str(root))
dest = preview.create_new_file("sub/new.md", "# hello")
assert dest == str(root / "sub" / "new.md")
assert (root / "sub" / "new.md").read_text(encoding="utf-8") == "# hello"
assert preview.current_file == dest # opened it, kept the AI chat (reset_ai=False)
def test_create_new_file_rejects_a_path_escaping_the_root(qt_app, root):
from cowork_local.presentation.folder.document_preview_manager import DocumentPreviewManager
preview = DocumentPreviewManager(str(root))
messages = []
preview.status_message.connect(messages.append)
dest = preview.create_new_file("../escaped.txt", "nope")
assert dest is None
assert not (root.parent / "escaped.txt").exists()
assert messages # a save_error status was emitted
def test_write_content_persists_and_refreshes_html_preview(qt_app, root):
from cowork_local.presentation.folder.document_preview_manager import DocumentPreviewManager
f = root / "page.html"
f.write_text("<p>old</p>", encoding="utf-8")
preview = DocumentPreviewManager(str(root))
preview.open_file(str(f)) # HTML opens in preview mode by default
preview.write_content("<p>new</p>")
assert f.read_text(encoding="utf-8") == "<p>new</p>"
# ---- FolderTab shell -------------------------------------------------------- #
def test_folder_tab_builds_and_wires_tree_to_preview(ctx, root):
from cowork_local.presentation.folder.folder_tab import FolderTab
tab = FolderTab(ctx)
tab.set_root(str(root))
f = root / "hello.py"
f.write_text("x = 1\n", encoding="utf-8")
tab.tree.file_selected.emit(str(f)) # simulate a tree click
assert tab.preview.current_file == str(f)
assert tab.preview.editor.toPlainText() == "x = 1\n"
def test_folder_tab_ai_panel_toggle_updates_button_badge(ctx, root):
from cowork_local.presentation.folder.folder_tab import FolderTab
tab = FolderTab(ctx)
assert tab.ai_panel.isHidden()
tab.ai_btn.setChecked(True)
tab._toggle_ai_panel()
assert not tab.ai_panel.isHidden()
@@ -0,0 +1,53 @@
"""EPIC R07-T03: QtSchedulerClock against a REAL QTimer/event loop.
``tests/unit/test_task_scheduler_dispatch.py`` covers ``TaskScheduler``'s
dispatch logic entirely through ``tests/fakes/fake_clock.py::FakeClock`` (no
Qt at all — that's the whole point of the extraction). This file is the
complement: it proves the adapter itself actually drives a real ``QTimer``
and pumps a real event loop, offscreen, the way ``test_history_dir_race.py``
proves ``ui/chat_panel.py``'s fix against real Qt rather than a double.
"""
from __future__ import annotations
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest # noqa: E402
from PySide6.QtTest import QTest # noqa: E402
from PySide6.QtWidgets import QApplication # noqa: E402
from cowork_local.infrastructure.qt.qt_scheduler_clock import QtSchedulerClock # noqa: E402
@pytest.fixture(scope="module")
def qapp():
return QApplication.instance() or QApplication([])
def test_start_fires_callback_on_the_real_qt_event_loop(qapp):
clock = QtSchedulerClock()
ticks = []
clock.start(interval_ms=10, callback=lambda: ticks.append(1))
try:
QTest.qWait(200) # let the real QTimer fire a few times
finally:
clock.stop()
assert len(ticks) >= 1
def test_stop_prevents_further_callbacks(qapp):
clock = QtSchedulerClock()
ticks = []
clock.start(interval_ms=10, callback=lambda: ticks.append(1))
QTest.qWait(50)
clock.stop()
count_after_stop = len(ticks)
QTest.qWait(100)
assert len(ticks) == count_after_stop # no more callbacks after stop()
def test_pump_processes_pending_events_without_raising(qapp):
clock = QtSchedulerClock()
clock.pump() # must not raise even with nothing pending
+125
View File
@@ -0,0 +1,125 @@
"""EPIC R08-T11: ScheduleTaskTab shell + KanbanBoardWidget, real Qt offscreen.
Drives the real widgets end to end (build -> refresh -> drag-drop rule via
TaskApplicationService -> refresh) against a tmp_path task repository, the
way ``test_history_dir_race.py`` proves R06-T04 against real Qt rather than
a double. ``TaskApplicationService``'s own business rules are already unit
tested (R07-T04); this file exists to prove the WIDGET is actually wired to
that service, not to re-test the rules themselves.
"""
from __future__ import annotations
import os
from pathlib import Path
import pytest
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from cowork_local.config import AppConfig # noqa: E402
from cowork_local.infrastructure.persistence.json.task_repository_impl import ( # noqa: E402
TaskRepository,
)
from cowork_local.state import AppContext # noqa: E402
pytest.importorskip("PySide6", reason="Qt is required for the integration suite")
@pytest.fixture(scope="module")
def qt_app():
from PySide6.QtWidgets import QApplication
return QApplication.instance() or QApplication([])
@pytest.fixture
def ctx(qt_app, tmp_path: Path):
return AppContext(AppConfig.load(tmp_path / "config.json"))
@pytest.fixture
def tasks_dir(tmp_path: Path) -> Path:
d = tmp_path / "tasks"
d.mkdir()
return d
def test_schedule_task_tab_builds_and_refreshes_with_no_tasks(ctx, tasks_dir):
from cowork_local.presentation.scheduling.schedule_task_tab import ScheduleTaskTab
tab = ScheduleTaskTab(ctx, scheduler=None, tasks_dir=tasks_dir)
tab.refresh() # must not raise against an empty repo
assert tab.kanban.columns.keys() # 7 lanes were built
def test_kanban_renders_a_task_into_its_status_lane(ctx, tasks_dir):
from cowork_local.presentation.scheduling.kanban_board_widget import KanbanBoardWidget
repo = TaskRepository(tasks_dir)
task = repo.create("My Task", task_type="cowork")
task["status"] = "backlog"
repo.save(task)
board = KanbanBoardWidget(ctx, tasks_dir=tasks_dir)
board.refresh()
from PySide6.QtCore import Qt
backlog_ids = [board.columns["backlog"].item(i).data(Qt.UserRole)
for i in range(board.columns["backlog"].count())]
assert task["task_id"] in backlog_ids
def test_dropping_a_card_on_done_disables_its_schedule_through_the_real_widget(ctx, tasks_dir):
"""Same rule TaskApplicationService.move_to_status covers at the unit
level (R07-T04) — this proves the Kanban widget's drop handler actually
calls it, end to end, with a real TaskRepository on disk."""
from cowork_local.presentation.scheduling.kanban_board_widget import KanbanBoardWidget
repo = TaskRepository(tasks_dir)
task = repo.create("Recurring", task_type="cowork")
task["schedule"]["enabled"] = True
task["schedule"]["run_at"] = "2026-08-28 09:00"
task["schedule"]["repeat_type"] = "daily"
task["status"] = "scheduled"
repo.save(task)
board = KanbanBoardWidget(ctx, tasks_dir=tasks_dir)
board.refresh()
board._on_task_dropped(task["task_id"], "done")
on_disk = repo.get(task["task_id"])
assert on_disk["status"] == "done"
assert on_disk["schedule"]["enabled"] is False
def test_kanban_edit_requested_is_wired_to_the_shells_edit_task(ctx, tasks_dir, monkeypatch):
"""Proves ScheduleTaskTab actually connects
``kanban.edit_requested -> self._edit_task`` (not just that the Kanban
widget emits the signal in isolation) by monkeypatching the dialog class
``_edit_task`` opens and checking it was constructed for the right task."""
import cowork_local.ui.task_editor_dialog as task_editor_dialog_module
from cowork_local.presentation.scheduling.schedule_task_tab import ScheduleTaskTab
repo = TaskRepository(tasks_dir)
task = repo.create("Editable")
repo.save(task)
seen_task_ids = []
class _FakeDialog:
def __init__(self, task, all_tasks, parent, ctx):
seen_task_ids.append(task["task_id"] if task else None)
self.edited_task = None
def exec(self):
return False # Cancel — nothing further should happen
monkeypatch.setattr(task_editor_dialog_module, "TaskEditorDialog", _FakeDialog)
tab = ScheduleTaskTab(ctx, scheduler=None, tasks_dir=tasks_dir)
tab.kanban.edit_requested.emit(task["task_id"])
assert seen_task_ids == [task["task_id"]]
@@ -0,0 +1,162 @@
"""EPIC R08-T14: GraphRenderer / GraphQaWidget / StructureGraphView shell,
real Qt offscreen.
Scope note: a real directory SCAN (``GraphRenderer._scan``) runs on a
``core.worker.AgentWorker`` QThread and had no existing test before this
task either (grep confirms nothing under tests/ exercised
``ui/structure_graph_view.py``). These tests drive ``_render()`` directly
with a hand-built ``StructureGraph`` instead of a live scan — enough to
prove the renderer <-> Q&A wiring (the actual R08-T14 deliverable) without
needing a real codebase to walk.
"""
from __future__ import annotations
import os
import pytest
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from cowork_local.config import AppConfig # noqa: E402
from cowork_local.core.structure_graph import GEdge, GNode, StructureGraph # noqa: E402
from cowork_local.state import AppContext # noqa: E402
pytest.importorskip("PySide6", reason="Qt is required for the integration suite")
@pytest.fixture(scope="module")
def qt_app():
from PySide6.QtWidgets import QApplication
return QApplication.instance() or QApplication([])
@pytest.fixture
def ctx(qt_app, tmp_path):
return AppContext(AppConfig.load(tmp_path / "config.json"))
def _fake_graph(tmp_path):
f = tmp_path / "mod.py"
f.write_text("def hello():\n pass\n", encoding="utf-8")
nodes = [
GNode(id="n1", label="mod.py", kind="file", detail="a module", path=str(f)),
GNode(id="n2", label="hello", kind="function", detail="says hi", path=str(f)),
]
edges = [GEdge(source="n1", target="n2", type="defines")]
return StructureGraph(nodes=nodes, edges=edges)
def test_structure_graph_view_builds(ctx):
from cowork_local.presentation.graph.structure_graph_view import StructureGraphView
view = StructureGraphView(ctx)
assert view.renderer is not None
assert view.qa is not None
def test_render_populates_the_scene_and_emits_graph_rendered(ctx, tmp_path):
from cowork_local.presentation.graph.graph_renderer import GraphRenderer
renderer = GraphRenderer(ctx)
graph = _fake_graph(tmp_path)
seen = []
renderer.graph_rendered.connect(lambda: seen.append(1))
renderer._render({"graph": graph, "pos": {"n1": (0, 0), "n2": (100, 0)}, "seq": renderer._scan_seq})
assert renderer.graph is graph
assert len(renderer._node_items) == 2
assert seen == [1]
def test_render_ignores_a_stale_scan_result(ctx, tmp_path):
"""Only the LATEST scan's result is ever rendered (no stale overwrite)."""
from cowork_local.presentation.graph.graph_renderer import GraphRenderer
renderer = GraphRenderer(ctx)
first = _fake_graph(tmp_path)
renderer._render({"graph": first, "pos": {}, "seq": renderer._scan_seq})
renderer._scan_seq += 1 # a second scan started
stale = StructureGraph(nodes=[], edges=[])
renderer._render({"graph": stale, "pos": {}, "seq": renderer._scan_seq - 1})
assert renderer.graph is first # stale result was dropped
def test_node_selection_updates_the_qa_detail_panel(ctx, tmp_path):
from cowork_local.presentation.graph.graph_qa_widget import GraphQaWidget
from cowork_local.presentation.graph.graph_renderer import GraphRenderer
renderer = GraphRenderer(ctx)
graph = _fake_graph(tmp_path)
renderer._render({"graph": graph, "pos": {"n1": (0, 0), "n2": (100, 0)}, "seq": renderer._scan_seq})
qa = GraphQaWidget(ctx, renderer)
node = graph.nodes[1]
renderer.node_selected.emit(node)
assert "hello" in qa.detail.toPlainText()
assert "says hi" in qa.detail.toPlainText()
def test_candidate_file_paths_uses_selected_nodes_when_present(ctx, tmp_path):
from cowork_local.presentation.graph.graph_qa_widget import GraphQaWidget
from cowork_local.presentation.graph.graph_renderer import GraphRenderer
renderer = GraphRenderer(ctx)
graph = _fake_graph(tmp_path)
renderer._render({"graph": graph, "pos": {"n1": (0, 0), "n2": (100, 0)}, "seq": renderer._scan_seq})
qa = GraphQaWidget(ctx, renderer)
# No selection -> every file node in the graph (both nodes share one
# file here, so this also proves the path-dedup in _candidate_file_paths).
assert qa._candidate_file_paths() == [graph.nodes[0].path]
renderer._node_items[0].setSelected(True)
assert qa._candidate_file_paths() == [graph.nodes[0].path]
def test_project_change_clears_the_qa_extraction_cache(ctx, tmp_path):
from cowork_local.presentation.graph.graph_qa_widget import GraphQaWidget
from cowork_local.presentation.graph.graph_renderer import GraphRenderer
renderer = GraphRenderer(ctx)
qa = GraphQaWidget(ctx, renderer)
qa._extract_cache = {"some/path.py": "cached text"}
renderer.project_changed.emit()
assert qa._extract_cache == {}
def test_qa_collapse_resizes_the_shells_splitter(ctx):
"""Exact pixel width is Qt's splitter-layout arithmetic, not this code's
concern - what matters is that collapsing narrows the QA pane a lot
(down near the strip's width) while the strip itself becomes visible and
the total splitter width is conserved."""
from cowork_local.presentation.graph.structure_graph_view import StructureGraphView
from cowork_local.ui.widgets import CollapseStrip
view = StructureGraphView(ctx)
before = view._split.sizes()
view.qa._set_collapsed(True)
after = view._split.sizes()
assert after[1] <= CollapseStrip.WIDTH + 2
assert view.qa.maximumWidth() == CollapseStrip.WIDTH + 2
assert sum(after) == sum(before) # total width conserved, just redistributed
def test_hide_event_clears_extracts(ctx):
from cowork_local.presentation.graph.structure_graph_view import StructureGraphView
view = StructureGraphView(ctx)
view.qa._extract_cache = {"x": "y"}
from PySide6.QtGui import QHideEvent
view.hideEvent(QHideEvent())
assert view.qa._extract_cache == {}