"""Tests for service project-members write (API key only, no acting).""" from __future__ import annotations import os from contextlib import contextmanager from unittest.mock import MagicMock import app.main # noqa: F401 import pytest from fastapi.testclient import TestClient from app.main import app from app.project_members_write import ProjectMemberWrite, 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 __enter__(self): return self def __exit__(self, *args): return False def _sample_item(**overrides): base = { "member_id": 10, "emp_id": 46, "emp_name": "Test Emp", "project_id": 86, "project_code": "P-86", "project_name": "Demo", "step_name": "РД", "status": 1, "status_name": "В работе", "archive": 0, "archive_date": None, "section_id": 12, "section_name": "ОВ", "role": 22, "active": 1, } base.update(overrides) return base @contextmanager def _fake_conn(cur): c = MagicMock() c.cursor.return_value = cur c.commit = MagicMock() yield c def test_validate_member_write_invalid_role(): cur = _FakeCursor() body = ProjectMemberWrite( project_id=1, emp_id=2, section_id=3, role=999, active=True ) with pytest.raises(Exception) as ei: validate_member_write(cur, "db", body) assert getattr(ei.value, "status_code", None) == 400 assert ei.value.detail["code"] == "invalid_role" def test_service_put_insert_without_acting(monkeypatch): import app.project_members_service as pms cur = _FakeCursor() monkeypatch.setattr(pms, "get_conn", lambda: _fake_conn(cur)) monkeypatch.setattr(pms, "_db_name", lambda _c: "db") monkeypatch.setattr( pms, "validate_member_write", lambda *_a, **_k: ({"team": 7, "id": 86}, [12]), ) monkeypatch.setattr(pms, "upsert_team_member", lambda *_a, **_k: 10) monkeypatch.setattr(pms, "fetch_member_item", lambda *_a, **_k: _sample_item()) prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "test-key" try: client = TestClient(app) r = client.put( "/api/service/project-members", headers={"X-Api-Key": "test-key"}, json={ "project_id": 86, "emp_id": 46, "section_id": 12, "role": 22, "active": True, }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 200 data = r.json() assert data["ok"] is True assert data["member_id"] == 10 assert data["emp_id"] == 46 assert data["section_id"] == 12 def test_service_put_update_without_acting(monkeypatch): import app.project_members_service as pms cur = _FakeCursor() monkeypatch.setattr(pms, "get_conn", lambda: _fake_conn(cur)) monkeypatch.setattr(pms, "_db_name", lambda _c: "db") monkeypatch.setattr( pms, "validate_member_write", lambda *_a, **_k: ({"team": 7, "id": 86}, [12]), ) monkeypatch.setattr(pms, "upsert_team_member", lambda *_a, **_k: 10) monkeypatch.setattr( pms, "fetch_member_item", lambda *_a, **_k: _sample_item(role=2, section_id=15, section_name="ТХ"), ) prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "test-key" try: client = TestClient(app) r = client.put( "/api/service/project-members", headers={"X-Api-Key": "test-key"}, json={ "project_id": 86, "emp_id": 46, "section_id": 15, "role": 2, "active": True, "text": "updated", }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 200 data = r.json() assert data["ok"] is True assert data["role"] == 2 assert data["section_id"] == 15 def test_service_put_validation_invalid_section(monkeypatch): import app.project_members_service as pms from fastapi import HTTPException cur = _FakeCursor() monkeypatch.setattr(pms, "get_conn", lambda: _fake_conn(cur)) monkeypatch.setattr(pms, "_db_name", lambda _c: "db") def boom(*_a, **_k): raise HTTPException( status_code=400, detail={"code": "invalid_section", "message": "Раздел не входит в состав проекта"}, ) monkeypatch.setattr(pms, "validate_member_write", boom) prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "test-key" try: client = TestClient(app) r = client.put( "/api/service/project-members", headers={"X-Api-Key": "test-key"}, json={ "project_id": 86, "emp_id": 46, "section_id": 999, "role": 22, }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 400 assert r.json()["detail"]["code"] == "invalid_section" def test_service_put_validation_archived_emp(monkeypatch): import app.project_members_service as pms from fastapi import HTTPException cur = _FakeCursor() monkeypatch.setattr(pms, "get_conn", lambda: _fake_conn(cur)) monkeypatch.setattr(pms, "_db_name", lambda _c: "db") def boom(*_a, **_k): raise HTTPException( status_code=400, detail={ "code": "invalid_employee", "message": "Нельзя добавить архивного или удалённого сотрудника", }, ) monkeypatch.setattr(pms, "validate_member_write", boom) prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "test-key" try: client = TestClient(app) r = client.put( "/api/service/project-members", headers={"X-Api-Key": "test-key"}, json={ "project_id": 86, "emp_id": 46, "section_id": 12, "role": 22, }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 400 assert r.json()["detail"]["code"] == "invalid_employee" def test_service_put_unknown_project(monkeypatch): import app.project_members_service as pms from fastapi import HTTPException cur = _FakeCursor() monkeypatch.setattr(pms, "get_conn", lambda: _fake_conn(cur)) monkeypatch.setattr(pms, "_db_name", lambda _c: "db") def boom(*_a, **_k): raise HTTPException( status_code=404, detail={"code": "project_not_found", "message": "Проект не найден"}, ) monkeypatch.setattr(pms, "validate_member_write", boom) prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "test-key" try: client = TestClient(app) r = client.put( "/api/service/project-members", headers={"X-Api-Key": "test-key"}, json={ "project_id": 99999, "emp_id": 46, "section_id": 12, "role": 22, }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 404 assert r.json()["detail"]["code"] == "project_not_found" def test_service_put_unknown_emp(monkeypatch): import app.project_members_service as pms from fastapi import HTTPException cur = _FakeCursor() monkeypatch.setattr(pms, "get_conn", lambda: _fake_conn(cur)) monkeypatch.setattr(pms, "_db_name", lambda _c: "db") def boom(*_a, **_k): raise HTTPException( status_code=404, detail={"code": "emp_not_found", "message": "Сотрудник 999 не найден"}, ) monkeypatch.setattr(pms, "validate_member_write", boom) prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "test-key" try: client = TestClient(app) r = client.put( "/api/service/project-members", headers={"X-Api-Key": "test-key"}, json={ "project_id": 86, "emp_id": 999, "section_id": 12, "role": 22, }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 404 assert r.json()["detail"]["code"] == "emp_not_found" def test_service_put_invalid_role_via_api(monkeypatch): """Pydantic rejects role < 1; invalid known-range role goes through validate.""" import app.project_members_service as pms from fastapi import HTTPException cur = _FakeCursor() monkeypatch.setattr(pms, "get_conn", lambda: _fake_conn(cur)) monkeypatch.setattr(pms, "_db_name", lambda _c: "db") def boom(*_a, **_k): raise HTTPException( status_code=400, detail={"code": "invalid_role", "message": "Укажите роль в проекте"}, ) monkeypatch.setattr(pms, "validate_member_write", boom) prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "test-key" try: client = TestClient(app) r = client.put( "/api/service/project-members", headers={"X-Api-Key": "test-key"}, json={ "project_id": 86, "emp_id": 46, "section_id": 12, "role": 999, }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 400 assert r.json()["detail"]["code"] == "invalid_role" def test_service_put_requires_api_key(): prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "secret-key" try: client = TestClient(app) r = client.put( "/api/service/project-members", json={ "project_id": 86, "emp_id": 46, "section_id": 12, "role": 22, }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 401 def test_portal_put_still_requires_acting(): prev = os.environ.get("USER_READER_API_KEY") os.environ["USER_READER_API_KEY"] = "test-key" try: client = TestClient(app) r = client.put( "/api/project-members", headers={"X-Api-Key": "test-key"}, json={ "project_id": 86, "emp_id": 46, "section_id": 12, "role": 22, }, ) finally: if prev is None: os.environ.pop("USER_READER_API_KEY", None) else: os.environ["USER_READER_API_KEY"] = prev assert r.status_code == 400 detail = r.json()["detail"] if isinstance(detail, dict): assert detail.get("code") == "missing_acting_emp" else: assert "Acting" in str(detail) or "acting" in str(detail).lower()