test(ttp): E.2.13 repository tests — TTPMixin idempotency + identity-rollup projection on dual backends

Adds tests/web/db/conftest.py with a db_backends fixture
parametrizing SQLite (always) + MySQL (gated on
DECNET_TEST_MYSQL_URL). Surface assertions (mixin methods present
+ async) GREEN today; insert_tags idempotency, identity rollup
projection, attacker-rollup exclusion of NULL-attacker tags
xfail-gated behind E.3.3.
This commit is contained in:
2026-05-01 07:39:16 -04:00
parent 6814949bc0
commit 4a93e16407
3 changed files with 206 additions and 0 deletions

0
tests/web/db/__init__.py Normal file
View File

71
tests/web/db/conftest.py Normal file
View File

@@ -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

View File

@@ -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")