test: add unit coverage and enforce CI gates

Author: izzzzzi

.github/workflows/ci.yml

@@ -13,6 +13,17 @@ jobs:
       - uses: actions/checkout@v4
       - uses: astral-sh/ruff-action@v3
 
+  test-python:
+    name: Unit Tests (Python)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - shell: bash
+        run: python -m unittest discover -s tests -p 'test_*.py'
+
   typecheck-ts:
     name: TypeScript (${{ matrix.os }})
     runs-on: ${{ matrix.os }}
@@ -26,3 +37,14 @@ jobs:
         run: cd ui && bun install --frozen-lockfile 2>/dev/null || bun install
       - shell: bash
         run: cd ui && bun run typecheck
+
+  test-ts:
+    name: Unit Tests (TypeScript)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: oven-sh/setup-bun@v2
+      - shell: bash
+        run: cd ui && bun install --frozen-lockfile 2>/dev/null || bun install
+      - shell: bash
+        run: cd ui && bun run test

.github/workflows/release.yml

@@ -14,9 +14,21 @@ jobs:
     name: Precheck
     uses: ./.github/workflows/ci.yml
 
+  package:
+    name: Package Check
+    needs: ci
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 22
+      - name: Verify npm package can be packed
+        run: npm pack --dry-run
+
   release:
     name: Semantic Release
-    needs: ci
+    needs: [ci, package]
     runs-on: ubuntu-latest
     if: github.repository == 'izzzzzi/izTeamSlots'
 

CONTRIBUTING.md

@@ -42,19 +42,30 @@ izTeamSlots/
 └── .github/workflows/ # CI/CD
 ```
 
-## Линтеры
+## Проверки и тесты
 
 Перед коммитом убедитесь что код проходит проверки:
 
 ```bash
-# Python — ruff
-ruff check backend/
+# Python — lint + unit tests
+ruff check backend tests
+python -m unittest discover -s tests -p 'test_*.py'
 
-# TypeScript — tsc
-npm run typecheck:ui
+# TypeScript — typecheck + unit tests
+npm --prefix ui run typecheck
+npm --prefix ui run test
 ```
 
-CI автоматически запускает оба линтера на каждый push и PR.
+CI автоматически запускает эти проверки на каждый PR.
+
+## Требование по тестам
+
+- Любая новая функциональность или заметное изменение логики должно сопровождаться тестами.
+- Минимум: покрывайте тот слой, который можно проверить локально без браузерного e2e.
+- Для Python-логики добавляйте `unittest`-тесты в `tests/`.
+- Для чистой TypeScript-логики добавляйте тесты в `ui/tests/`.
+- Если изменение нельзя адекватно покрыть unit-тестом, это нужно явно отметить в описании PR.
+- PR без тестов для новой логики может быть отклонён.
 
 ## Conventional Commits
 
@@ -100,8 +111,9 @@ chore: update seleniumbase to 4.33
 
 1. Один PR — одно логическое изменение.
 2. Следуйте Conventional Commits.
-3. Убедитесь что `ruff check` и `tsc --noEmit` проходят.
-4. Обновите документацию если изменение затрагивает пользовательское поведение.
+3. Добавьте или обновите тесты, если меняется логика приложения.
+4. Убедитесь что lint, unit tests и typecheck проходят локально.
+5. Обновите документацию если изменение затрагивает пользовательское поведение.
 
 ### Новый почтовый провайдер
 

backend/rpc_server.py

@@ -155,6 +155,19 @@ def _handle_request(self, req: RPCRequest) -> dict[str, Any]:
             )
             return make_success_response(req.request_id, {"job_id": job_id})
 
+        if m == "workspace.sync_preview":
+            admin_email = self._as_str_param(p, "admin_email")
+            result = self.facade.preview_workspace_sync(admin_email)
+            return make_success_response(req.request_id, result)
+
+        if m == "job.sync_workspace":
+            admin_email = self._as_str_param(p, "admin_email")
+            job_id = self._run_job(
+                f"Синхронизация WS: {admin_email}",
+                lambda ctx: self.facade.sync_workspace(admin_email, ctx.log),
+            )
+            return make_success_response(req.request_id, {"job_id": job_id})
+
         if m == "job.open_admin_browser":
             email = self._as_str_param(p, "email")
             job_id = self._run_job(

backend/slot_orchestrator.py

@@ -310,6 +310,106 @@ def get_members(self) -> list[dict]:
         api = self._get_api(page)
         return api.get_members()
 
+    def _build_workspace_sync_plan(
+        self,
+        members: list[dict[str, Any]],
+        pending_invites: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        local_emails = {w.email.strip().lower() for w in self._get_workers() if w.email}
+        admin_email = (self.admin_email or "").strip().lower()
+
+        extra_members: list[dict[str, str]] = []
+        extra_invites: list[dict[str, str]] = []
+        skipped: list[dict[str, str]] = []
+
+        for member in members:
+            email = str(member.get("email") or "").strip()
+            if not email:
+                continue
+            email_key = email.lower()
+            role = str(member.get("role") or "")
+
+            if email_key == admin_email:
+                skipped.append({"type": "member", "email": email, "reason": "self"})
+                continue
+            if role and role != "standard-user":
+                skipped.append({"type": "member", "email": email, "reason": f"role:{role}"})
+                continue
+            if email_key in local_emails:
+                continue
+
+            extra_members.append(
+                {
+                    "email": email,
+                    "id": str(member.get("id") or ""),
+                    "role": role or "standard-user",
+                }
+            )
+
+        for invite in pending_invites:
+            email = str(invite.get("email") or invite.get("email_address") or "").strip()
+            if not email:
+                continue
+            email_key = email.lower()
+            if email_key == admin_email:
+                skipped.append({"type": "invite", "email": email, "reason": "self"})
+                continue
+            if email_key in local_emails:
+                continue
+
+            extra_invites.append({"email": email})
+
+        return {
+            "admin_email": self.admin_email,
+            "workspace_id": self.account_id,
+            "local_workers_total": len(local_emails),
+            "members_total": len(members),
+            "pending_invites_total": len(pending_invites),
+            "extra_members": extra_members,
+            "extra_invites": extra_invites,
+            "skipped": skipped,
+            "removed_members": [],
+            "removed_invites": [],
+        }
+
+    def sync_workspace(self, *, dry_run: bool = True) -> dict[str, Any]:
+        page = self._ensure_admin_page()
+        api = self._get_api(page)
+
+        members = api.get_members()
+        pending_invites = api.get_pending_invites()
+        result = self._build_workspace_sync_plan(members, pending_invites)
+        result["dry_run"] = dry_run
+
+        if dry_run:
+            self._log(f"WS preview: лишних участников {len(result['extra_members'])}, лишних инвайтов {len(result['extra_invites'])}")
+            return result
+
+        for member in result["extra_members"]:
+            member_id = member.get("id")
+            email = member.get("email", "")
+            if not member_id:
+                self._log(f"[предупреждение] Пропускаю участника без id: {email}")
+                continue
+            api.delete_member(member_id)
+            result["removed_members"].append(email)
+            self._log(f"Удалён участник WS: {email}")
+
+        for invite in result["extra_invites"]:
+            email = invite.get("email", "")
+            if not email:
+                continue
+            api.delete_invite(email)
+            result["removed_invites"].append(email)
+            self._log(f"Удалён инвайт WS: {email}")
+
+        self._log(
+            "WS синхронизирован: "
+            f"участников удалено {len(result['removed_members'])}, "
+            f"инвайтов удалено {len(result['removed_invites'])}"
+        )
+        return result
+
     def register_slot(self, worker: WorkerAccount, invite_url: str) -> None:
         """Зарегистрировать приглашённый аккаунт через инвайт-ссылку."""
         mailbox = self._mailboxes.get(worker.email)

backend/ui_facade.py

@@ -209,6 +209,18 @@ def login_admin_manual(self, email: str, log: LogFunc) -> None:
         manager.login_admin_manual()
         self.sync_codex_files()
 
+    def preview_workspace_sync(self, admin_email: str) -> dict[str, Any]:
+        manager = SlotManager(store=self.store, admin_email=admin_email, log=lambda _msg: None, headless=True)
+        try:
+            return manager.sync_workspace(dry_run=True)
+        finally:
+            manager.close()
+
+    def sync_workspace(self, admin_email: str, log: LogFunc) -> dict[str, Any]:
+        manager = SlotManager(store=self.store, admin_email=admin_email, log=log, headless=False)
+        self._replace_manager(manager)
+        return manager.sync_workspace(dry_run=False)
+
     def run_slots_pipeline(
         self,
         admin_email: str,

tests/test_account_store.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from backend.account_store import AccountStore
+
+
+class TestAccountStore(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.base_dir = Path(self.temp_dir.name) / "accounts"
+        self.store = AccountStore(self.base_dir)
+
+    def tearDown(self) -> None:
+        self.temp_dir.cleanup()
+
+    def test_admin_crud_roundtrip(self) -> None:
+        admin = self.store.add_admin("admin@example.com", "pw")
+        admin.access_token = "token"
+        admin.workspace_id = "ws-1"
+        admin.account_id = "acc-1"
+        admin.workspaces = [{"workspace_id": "ws-1"}]
+        admin.last_login = "2026-03-07T00:00:00Z"
+        self.store.update_admin(admin)
+
+        loaded = self.store.get_admin("admin@example.com")
+
+        self.assertIsNotNone(loaded)
+        assert loaded is not None
+        self.assertEqual(loaded.access_token, "token")
+        self.assertEqual(loaded.workspace_id, "ws-1")
+        self.assertEqual(loaded.account_id, "acc-1")
+        self.assertEqual(loaded.workspaces, [{"workspace_id": "ws-1"}])
+        self.assertEqual(loaded.last_login, "2026-03-07T00:00:00Z")
+
+        self.store.delete_admin("admin@example.com")
+        self.assertIsNone(self.store.get_admin("admin@example.com"))
+
+    def test_worker_crud_roundtrip(self) -> None:
+        worker = self.store.add_worker("slot@example.com", "pw", "admin@example.com")
+        worker.status = "registered"
+        worker.openai_password = "openai-pw"
+        worker.access_token = "token"
+        worker.workspace_id = "ws-1"
+        self.store.update_worker(worker)
+
+        loaded = self.store.get_worker("slot@example.com")
+
+        self.assertIsNotNone(loaded)
+        assert loaded is not None
+        self.assertEqual(loaded.status, "registered")
+        self.assertEqual(loaded.openai_password, "openai-pw")
+        self.assertEqual(loaded.access_token, "token")
+        self.assertEqual(loaded.workspace_id, "ws-1")
+        self.assertEqual(loaded.admin_email, "admin@example.com")
+
+        self.store.delete_worker("slot@example.com")
+        self.assertIsNone(self.store.get_worker("slot@example.com"))
+
+    def test_doctor_rebuilds_broken_worker_index(self) -> None:
+        worker = self.store.add_worker("slot@example.com", "pw", "admin@example.com")
+        worker_dir = self.store.worker_dir / worker.id
+        index_path = self.store.worker_dir / "index.json"
+        index_path.write_text("{broken", encoding="utf-8")
+
+        fixes = self.store.doctor()
+
+        rebuilt = json.loads(index_path.read_text(encoding="utf-8"))
+        self.assertIn("slot@example.com", rebuilt)
+        self.assertTrue(any("index.json был повреждён" in fix for fix in fixes))
+        self.assertTrue(worker_dir.exists())
+
+    def test_doctor_recreates_missing_meta(self) -> None:
+        worker = self.store.add_worker("slot@example.com", "pw", "admin@example.com")
+        meta_path = self.store.worker_dir / worker.id / "meta.json"
+        meta_path.unlink()
+
+        fixes = self.store.doctor()
+
+        rebuilt = json.loads(meta_path.read_text(encoding="utf-8"))
+        self.assertEqual(rebuilt["email"], "slot@example.com")
+        self.assertEqual(rebuilt["status"], "created")
+        self.assertTrue(any("meta.json отсутствовал" in fix for fix in fixes))

tests/test_dto.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import unittest
+
+from backend.account_store import AdminAccount, WorkerAccount
+from backend.dto import AdminRow, AppStateDTO, WorkerRow
+
+
+class TestDTO(unittest.TestCase):
+    def test_admin_row_statuses(self) -> None:
+        account = AdminAccount(id="1", email="admin@example.com", password="pw")
+        self.assertEqual(AdminRow.from_account(account).status_label, "Не настроен")
+        self.assertEqual(AdminRow.from_account(account, has_browser_profile=True).status_label, "Нужен вход")
+
+        account.access_token = "token"
+        self.assertEqual(AdminRow.from_account(account).status_label, "Есть токен")
+        self.assertEqual(AdminRow.from_account(account, has_browser_profile=True).status_label, "Готов")
+
+    def test_worker_row_statuses(self) -> None:
+        account = WorkerAccount(id="1", email="slot@example.com", password="pw", admin_email="admin@example.com")
+        self.assertEqual(WorkerRow.from_account(account).status_label, "Создан")
+
+        account.status = "invited"
+        self.assertEqual(WorkerRow.from_account(account).status_label, "Инвайт отправлен")
+
+        account.status = "registered"
+        self.assertEqual(WorkerRow.from_account(account).status_label, "Зарегистрирован")
+
+        account.access_token = "token"
+        self.assertEqual(WorkerRow.from_account(account, has_browser_profile=True).status_label, "Готов")
+
+    def test_app_state_to_dict(self) -> None:
+        dto = AppStateDTO(
+            admins=[AdminRow.from_account(AdminAccount(id="1", email="admin@example.com", password="pw"))],
+            workers=[WorkerRow.from_account(WorkerAccount(id="2", email="slot@example.com", password="pw"))],
+        )
+
+        result = dto.to_dict()
+
+        self.assertIn("admins", result)
+        self.assertIn("workers", result)
+        self.assertEqual(result["admins"][0]["email"], "admin@example.com")
+        self.assertEqual(result["workers"][0]["email"], "slot@example.com")