stack/tests/pfs/test_codetables.py

"""pfs.codetables — DDL + writers for the code element/event/family tables."""

from __future__ import annotations

import duckdb
import pytest

from pfs.codetables import (
    ElementRow,
    EventRow,
    FamilyRow,
    ReviewRow,
    ensure_tables,
    read_all_elements,
    read_all_events,
    read_elements,
    read_events,
    read_families,
    write_elements,
    write_events,
    write_families,
)


@pytest.fixture
def con():
    c = duckdb.connect(":memory:")
    ensure_tables(c)
    yield c
    c.close()


def _el(code="99490", value="consent", year=2015):
    return ElementRow(
        code, year, "activity", value, "", "Consent;", "DE2VH9PD", 1245, 67716, "fr"
    )


class TestDDL:
    def test_idempotent(self, con):
        ensure_tables(con)
        names = {
            r[0]
            for r in con.execute(
                "SELECT table_name FROM information_schema.tables WHERE table_schema='pfs'"
            ).fetchall()
        }
        assert {
            "code_element",
            "code_element_review",
            "code_event",
            "code_family",
        } <= names


class TestElements:
    def test_write_replaces_per_code(self, con):
        assert (
            write_elements(
                con,
                "99490",
                [_el(), _el(value="24-7-access")],
                [ReviewRow("99490", "odd line", "activity", "", "DE2VH9PD", 1246)],
            )
            == 2
        )
        assert write_elements(con, "99490", [_el()], []) == 1
        rows = read_elements(con, "99490")
        assert [r.value for r in rows] == ["consent"]
        assert (
            con.execute(
                "SELECT count(*) FROM pfs.code_element_review WHERE code='99490'"
            ).fetchone()[0]
            == 0
        )

    def test_other_codes_untouched(self, con):
        write_elements(con, "99490", [_el()], [])
        write_elements(con, "G0556", [_el(code="G0556")], [])
        write_elements(con, "99490", [], [])
        assert read_elements(con, "99490") == []
        assert len(read_elements(con, "G0556")) == 1


class TestReadAll:
    def test_read_all_elements_matches_per_code_reader(self, con):
        write_elements(
            con, "99490", [_el(value="24-7-access"), _el(value="consent")], []
        )
        write_elements(con, "G0556", [_el(code="G0556")], [])
        out = read_all_elements(con)
        assert set(out) == {"99490", "G0556"}
        assert out["99490"] == read_elements(con, "99490")
        assert out["G0556"] == read_elements(con, "G0556")

    def test_read_all_elements_empty(self, con):
        assert read_all_elements(con) == {}

    def test_read_all_events_matches_per_code_reader(self, con):
        rows = [
            EventRow(
                "99439",
                2021,
                "replaces",
                "G2058",
                "99439",
                "YBM4IZUS",
                1578,
                84639,
                "fr",
                True,
                "",
            ),
            EventRow("99439", 2021, "appeared", "", "", "", 0, 0, "rvu", True, ""),
        ]
        write_events(con, "99439", rows)
        write_events(
            con,
            "G2058",
            [EventRow("G2058", 2020, "appeared", "", "", "", 0, 0, "rvu", True, "")],
        )
        out = read_all_events(con)
        assert set(out) == {"99439", "G2058"}
        assert out["99439"] == read_events(con, "99439")
        assert out["G2058"] == read_events(con, "G2058")

    def test_read_all_events_empty(self, con):
        assert read_all_events(con) == {}


class TestEvents:
    def test_roundtrip_sorted_by_year(self, con):
        rows = [
            EventRow(
                "99439",
                2021,
                "replaces",
                "G2058",
                "99439",
                "YBM4IZUS",
                1578,
                84639,
                "fr",
                True,
                "",
            ),
            EventRow("99439", 2021, "appeared", "", "", "", 0, 0, "rvu", True, ""),
        ]
        assert write_events(con, "99439", rows) == 2
        got = read_events(con, "99439")
        assert [e.kind for e in got] == ["appeared", "replaces"]
        assert got[1].anchored is True


class TestFamilies:
    def test_full_replace(self, con):
        write_families(
            con,
            [
                FamilyRow(
                    "CCM", "Chronic Care Management", "99490", "base", 2015, None, "", 0
                )
            ],
        )
        write_families(
            con,
            [
                FamilyRow(
                    "CCM",
                    "Chronic Care Management",
                    "99439",
                    "add-on",
                    2021,
                    None,
                    "YBM4IZUS",
                    1578,
                )
            ],
        )
        fams = read_families(con)
        assert len(fams) == 1 and fams[0].code == "99439" and fams[0].role == "add-on"