meraproject/services/user-reader/tests/test_project_members_write.py

"""Unit tests for project member write logic."""

import app.main  # noqa: F401 — register routers before submodule imports

from app.labor_permissions import evaluate_is_rw_member
from app.member_roles import is_valid_member_role, list_member_roles
from app.merakomis_schema import ROLE_MAIN_ENGINEER
from app.project_members_write import ProjectMemberWrite, upsert_team_member, validate_member_write


class _FakeCursor:
    def __init__(self, fetchone_results: list | None = None):
        self._fetchone = list(fetchone_results or [])
        self.executed: list[tuple] = []

    def execute(self, sql, params=None):
        self.executed.append((sql, params))

    def fetchone(self):
        if self._fetchone:
            return self._fetchone.pop(0)
        return None

    def fetchall(self):
        return []


def test_evaluate_is_rw_member_admin():
    assert evaluate_is_rw_member(
        is_admin_user=True,
        is_rp=False,
        acting_role=0,
        is_dept_director=False,
        is_edit=False,
        target_in_subemps=False,
    )


def test_evaluate_is_rw_member_rp():
    assert evaluate_is_rw_member(
        is_admin_user=False,
        is_rp=True,
        acting_role=0,
        is_dept_director=False,
        is_edit=False,
        target_in_subemps=False,
    )


def test_evaluate_is_rw_member_gip():
    assert evaluate_is_rw_member(
        is_admin_user=False,
        is_rp=False,
        acting_role=ROLE_MAIN_ENGINEER,
        is_dept_director=False,
        is_edit=False,
        target_in_subemps=False,
    )


def test_evaluate_is_rw_member_dept_director_create():
    assert evaluate_is_rw_member(
        is_admin_user=False,
        is_rp=False,
        acting_role=0,
        is_dept_director=True,
        is_edit=False,
        target_in_subemps=False,
    )


def test_evaluate_is_rw_member_dept_edit_subordinate():
    assert evaluate_is_rw_member(
        is_admin_user=False,
        is_rp=False,
        acting_role=0,
        is_dept_director=False,
        is_edit=True,
        target_in_subemps=True,
    )


def test_evaluate_is_rw_member_denied():
    assert not evaluate_is_rw_member(
        is_admin_user=False,
        is_rp=False,
        acting_role=22,
        is_dept_director=False,
        is_edit=True,
        target_in_subemps=False,
    )


def test_member_roles_list():
    roles = list_member_roles()
    assert any(r["id"] == 22 for r in roles)
    assert all(r["id"] != 8 for r in roles)


def test_member_roles_include_archived():
    roles = list_member_roles(include_role=8)
    assert any(r["id"] == 8 for r in roles)


def test_is_valid_member_role():
    assert is_valid_member_role(22)
    assert not is_valid_member_role(999)


def test_validate_member_write_invalid_role():
    cur = _FakeCursor()
    body = ProjectMemberWrite(
        project_id=1, emp_id=2, section_id=3, role=999, active=True
    )
    try:
        validate_member_write(cur, "db", body)
        assert False, "expected HTTPException"
    except Exception as e:
        assert getattr(e, "status_code", None) == 400
        assert e.detail["code"] == "invalid_role"


def test_upsert_paths(monkeypatch):
    cur = _FakeCursor([{"id": 5, "emp": 2, "team": 10}])
    cur.lastrowid = 99

    def fake_columns(_cur, _db):
        return {
            "table": "tMerakomisTeamMember",
            "id": "id",
            "emp": "emp",
            "team": "team",
            "section": "section",
            "role": "role",
            "active": "active",
            "text": "text",
            "portal": "portal",
        }

    import app.project_members_write as pmw

    monkeypatch.setattr(pmw, "_member_columns", fake_columns)
    monkeypatch.setattr(
        pmw,
        "fetch_team_member_row",
        lambda _c, _d, team, emp: {"id": 5, "emp": emp, "team": team},
    )

    member_id = upsert_team_member(
        cur,
        "db",
        team_id=10,
        emp_id=2,
        section_id=3,
        role=22,
        active=True,
        text="",
    )
    assert member_id == 5
    assert any("UPDATE" in sql for sql, _ in cur.executed)

    cur2 = _FakeCursor([None])
    cur2.lastrowid = 42
    monkeypatch.setattr(pmw, "fetch_team_member_row", lambda *_a, **_k: None)
    member_id2 = upsert_team_member(
        cur2,
        "db",
        team_id=10,
        emp_id=3,
        section_id=3,
        role=22,
        active=True,
        text="note",
    )
    assert member_id2 == 42
    assert any("INSERT" in sql for sql, _ in cur2.executed)