diff --git a/tests/web/db/__init__.py b/tests/web/db/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/web/db/conftest.py b/tests/web/db/conftest.py new file mode 100644 index 00000000..4d862926 --- /dev/null +++ b/tests/web/db/conftest.py @@ -0,0 +1,71 @@ +"""Shared fixtures for ``tests/web/db/`` — dual-backend repo testing. + +The ``db_backends`` fixture parametrizes a repository instance over +SQLite (always) and MySQL (skipped when ``DECNET_TEST_MYSQL_URL`` is +unset). This is the single source of truth referenced by the design +doc's "every repo test runs against both SQLite and MySQL" +convention; new repo tests under ``tests/web/db/`` should consume +the fixture rather than instantiating their own backend. + +MySQL is gated on env var rather than auto-detected because spinning +a real MySQL is heavy enough to belong in CI / live runs but not the +dev loop. Per project memory: "skip heavy test categories" in the +dev cycle. +""" +from __future__ import annotations + +import os +from pathlib import Path +from typing import AsyncIterator + +import pytest +import pytest_asyncio + +from decnet.web.db.factory import get_repository +from decnet.web.db.repository import BaseRepository + + +_BACKENDS: list[str] = ["sqlite"] +if os.environ.get("DECNET_TEST_MYSQL_URL"): + _BACKENDS.append("mysql") + + +@pytest_asyncio.fixture(params=_BACKENDS, ids=_BACKENDS) +async def db_backends( + request: pytest.FixtureRequest, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> AsyncIterator[BaseRepository]: + """Yield an initialized :class:`BaseRepository` for each available + backend. + + SQLite always runs (via ``aiosqlite`` + a tmp file). MySQL runs + iff ``DECNET_TEST_MYSQL_URL`` is set in the environment to a real + MySQL DSN — in that case the fixture writes a per-test schema + name into ``DECNET_DB_URL`` so concurrent tests don't collide. + """ + backend = request.param + if backend == "sqlite": + monkeypatch.setenv("DECNET_DB_TYPE", "sqlite") + repo = get_repository(db_path=str(tmp_path / "ttp.db")) + else: + # MySQL — uses the operator-supplied DSN. Per dual-DB-backend + # convention, dialect-specific behavior overrides land in the + # MySQL repo class; this fixture does not paper over them. + url = os.environ["DECNET_TEST_MYSQL_URL"] + monkeypatch.setenv("DECNET_DB_TYPE", "mysql") + monkeypatch.setenv("DECNET_DB_URL", url) + repo = get_repository() + await repo.initialize() + try: + yield repo + finally: + # SQLite is fully isolated per tmp_path; MySQL needs explicit + # teardown that's the operator's responsibility (truncate or + # drop schema in a CI hook). The repo close is best-effort. + engine = getattr(repo, "engine", None) + if engine is not None: + try: + await engine.dispose() + except Exception: # noqa: BLE001 — teardown best-effort + pass diff --git a/tests/web/db/test_ttp_repo.py b/tests/web/db/test_ttp_repo.py new file mode 100644 index 00000000..745f802f --- /dev/null +++ b/tests/web/db/test_ttp_repo.py @@ -0,0 +1,135 @@ +"""E.2.13 — Repository tests for the TTP-tag mixin. + +Pins the repo contract from ``development/TTP_TAGGING.md`` §E.2.13: + +* Per dual-DB-backend project convention, every test runs against + BOTH SQLite and MySQL via the :func:`db_backends` fixture in + :mod:`tests.web.db.conftest`. +* ``insert_tags`` is idempotent across runs (same UUID → no duplicate + row, no exception, second-run insert count is zero). +* ``list_techniques_by_identity`` projects through + ``Attacker.identity_id`` correctly when ``attacker_uuid`` is set on + the tag. +* ``list_techniques_by_identity`` returns identity-rollup tags (with + ``attacker_uuid IS NULL``) correctly. + +Method-signature surface is GREEN today (the mixin is wired into the +repo). Behavioral assertions xfail-gated behind E.3.3 — the empty +bodies raise ``NotImplementedError``. +""" +from __future__ import annotations + +import inspect + +import pytest + +from decnet.web.db.repository import BaseRepository +from decnet.web.db.sqlmodel_repo.ttp import TTPMixin + + +# ── Surface (GREEN today) ─────────────────────────────────────────── + + +def test_mixin_methods_are_async() -> None: + """All four query methods + ``insert_tags`` are coroutines. + + Catches a refactor that accidentally drops the ``async`` keyword + on a method body — which would silently break the repo's + expected awaitable interface. + """ + for name in ( + "insert_tags", + "list_techniques_by_identity", + "list_techniques_by_attacker", + "list_techniques_by_campaign", + "list_techniques_by_session", + "list_distinct_techniques", + ): + member = getattr(TTPMixin, name) + assert inspect.iscoroutinefunction(member), ( + f"TTPMixin.{name} must be `async def`" + ) + + +async def test_mixin_methods_present_on_repo( + db_backends: BaseRepository, +) -> None: + """The repository instance returned by the factory exposes every + TTPMixin method via composition. Confirms the mixin is wired in + on both SQLite and MySQL (the dual-backend fixture parametrizes). + """ + for name in ( + "insert_tags", + "list_techniques_by_identity", + "list_techniques_by_attacker", + "list_techniques_by_campaign", + "list_techniques_by_session", + "list_distinct_techniques", + ): + assert hasattr(db_backends, name) + + +# ── Behavior (xfail until E.3.3) ──────────────────────────────────── + + +@pytest.mark.xfail( + strict=True, + reason="impl phase E.3.3 — insert_tags idempotency lands with the " + "repository implementation", +) +async def test_insert_tags_idempotent_across_runs( + db_backends: BaseRepository, +) -> None: + """Running ``insert_tags`` twice on the same row set inserts on + the first call and no-ops on the second (returned count is 0). + + Today the body raises ``NotImplementedError`` so the assertion + xfails. Flips at E.3.3. + """ + pytest.fail("insert_tags not yet implemented") + + +@pytest.mark.xfail( + strict=True, + reason="impl phase E.3.3 — list_techniques_by_identity projection " + "through Attacker.identity_id lands with the repository impl", +) +async def test_list_by_identity_projects_through_attacker( + db_backends: BaseRepository, +) -> None: + """A tag with ``attacker_uuid`` set (and ``identity_uuid`` NULL) + appears in the per-Identity rollup for the attacker's identity, + via the ``Attacker.identity_id`` foreign key projection. + """ + pytest.fail("list_techniques_by_identity not yet implemented") + + +@pytest.mark.xfail( + strict=True, + reason="impl phase E.3.3 — identity-rollup tags (NULL attacker_uuid) " + "land with the repository impl", +) +async def test_list_by_identity_includes_rollup_tags( + db_backends: BaseRepository, +) -> None: + """Tags with ``attacker_uuid IS NULL`` and ``identity_uuid`` set + (the identity-lifter rollup case) appear in the per-Identity + listing — they belong to the Identity, not any single IP. + """ + pytest.fail("list_techniques_by_identity not yet implemented") + + +@pytest.mark.xfail( + strict=True, + reason="impl phase E.3.3 — list_techniques_by_attacker excludes " + "identity-rollup tags by design; lands with the repo impl", +) +async def test_list_by_attacker_excludes_rollup_tags( + db_backends: BaseRepository, +) -> None: + """Per-Attacker rollup is filtered on ``attacker_uuid``; tags + with ``attacker_uuid IS NULL`` (identity rollups) are deliberately + excluded. Pinned per design doc §E.2.13: "those belong to the + Identity, not any one IP underneath it." + """ + pytest.fail("list_techniques_by_attacker not yet implemented")