diff --git a/.github/workflows/release-pr.yml b/.github/workflows/release-pr.yml index 4aeb5b4..fd48afb 100644 --- a/.github/workflows/release-pr.yml +++ b/.github/workflows/release-pr.yml @@ -1,14 +1,22 @@ name: Create Release PR on: + push: + branches: [dev] workflow_dispatch: +concurrency: + group: release-pr + cancel-in-progress: false + permissions: contents: read pull-requests: write jobs: create-pr: + environment: + name: release-pr runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -24,6 +32,10 @@ jobs: pr_body: | Automated PR to merge `dev` into `master`. + This workflow runs on every push to `dev`, but job execution can + be gated by required reviewers on the `release-pr` environment. + That keeps PR creation automatic while still requiring approval. + --- _Created via Release PR workflow._ pr_label: "release" diff --git a/.gitignore b/.gitignore index c549e84..5a72ed6 100644 --- a/.gitignore +++ b/.gitignore @@ -59,10 +59,11 @@ Thumbs.db .pytest_cache/ .coverage htmlcov/ +tests/ # mypy .mypy_cache/ clean_* -target/ \ No newline at end of file +target/ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..6633053 --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,184 @@ +# Contributing + +Zero Ichi moves fast. Good contributions keep it stable while making it easier +to operate, extend, and debug. + +This guide is for people sending code, docs, or workflow changes. + +## Before You Start + +- Open an issue first for bigger changes, behavior changes, or new features. +- Keep one pull request focused on one problem. +- If a change affects commands, config, or user-facing text, update docs and + locales in same PR. + +## Local Setup + +```bash +git clone https://github.com/MhankBarBar/zero-ichi +cd zero-ichi +uv sync +cp .env.example .env +``` + +Run bot: + +```bash +uv run zero-ichi +``` + +Interactive first-run setup: + +```bash +uv run zero-ichi setup +``` + +Docs dev server: + +```bash +cd docs +bun install +bun run docs:dev +``` + +## Branching + +- Branch from `dev`. +- Use short branch names that say what changed. +- Examples: + - `fix/addskill-bom` + - `feat/privacy-controls` + - `docs/mobile-table-overflow` + +## Project Rules + +### Commands + +- Add commands under `src/commands//`. +- Follow `Command` base class pattern already used in repo. +- Respect existing permission flags: `owner_only`, `admin_only`, + `group_only`, `private_only`. +- Reuse shared helpers before adding new mini-frameworks. + +### Config Changes + +If you add or change runtime config: + +1. Update `DEFAULT_CONFIG` in `src/core/runtime_config.py` +2. Update `config.schema.json` +3. Preserve merge/backfill behavior for existing user config +4. Add or update command/docs examples if users need to touch it + +### User-Facing Text + +If command output changes: + +1. Add/update `src/locales/en.json` +2. Add/update `src/locales/id.json` +3. Keep keys parallel across both files + +### Docs + +Update docs when you change: + +- command behavior +- config shape +- workflow/release process +- setup flow +- moderation behavior + +Main docs live in `docs/`. + +## Validation + +Minimum before opening PR: + +```bash +uv run ruff format . +uv run ruff check . +``` + +Run tests relevant to your change. + +If you are preparing a merge-ready branch, run full suite too: + +```bash +uv run pytest -q +``` + +Docs changes should also build: + +```bash +cd docs +bun run docs:build +``` + +## Tests Policy + +Repo currently ignores `tests/` in Git. + +- Keep exploratory tests local. +- If you need permanent test coverage in-repo, discuss it in issue/PR first so + policy stays deliberate instead of accidental. + +## Pull Request Expectations + +Good PRs include: + +- clear title +- short summary of problem +- short summary of fix +- validation steps you ran +- screenshots or terminal output when UI/docs behavior changed + +Bad PRs usually have one of these problems: + +- mix refactor and feature work with no separation +- update code but not docs/locales/schema +- add duplicate helper instead of extending shared one +- change behavior without explaining why + +## Commit Messages + +Use Conventional Commits. + +Examples: + +```text +fix(ai): handle BOM in skill markdown +feat(config): add rollback history +docs(commands): clarify addskill usage +refactor(core): share prefixed id generator +``` + +## Release PR Automation + +Repo has release PR workflow for `dev -> master`. + +- Push to `dev` triggers release PR workflow automatically. +- Job uses `release-pr` environment. +- To require approval before PR creation, configure required reviewers in + GitHub Settings -> Environments -> `release-pr`. + +That gives automatic trigger with explicit approval gate. + +## What Usually Needs Extra Care + +- command permission logic +- runtime config persistence and schema validation +- WhatsApp message parsing, especially quoted/media messages +- AI flows that edit messages after async work +- docs tables/layout on mobile +- GitHub workflows that can create PRs, push branches, or deploy pages + +## Reporting Problems + +Open issue with: + +- exact command or workflow involved +- steps to reproduce +- expected result +- actual result +- logs, screenshots, or copied error text + +Short, concrete reports get fixed faster. diff --git a/docs/.vitepress/theme/custom.css b/docs/.vitepress/theme/custom.css index 6354987..2939e4e 100644 --- a/docs/.vitepress/theme/custom.css +++ b/docs/.vitepress/theme/custom.css @@ -1023,6 +1023,34 @@ select:focus { /* ─── Responsive ───────────────────────────────────────────────── */ @media (max-width: 960px) { + html, + body, + .Layout, + .VPDoc { + overflow-x: hidden; + } + + .vp-doc table { + display: block; + width: 100%; + max-width: 100%; + overflow-x: auto; + -webkit-overflow-scrolling: touch; + table-layout: auto; + } + + .vp-doc table thead, + .vp-doc table tbody { + display: table; + width: max-content; + min-width: 100%; + } + + .vp-doc th, + .vp-doc td { + white-space: nowrap; + } + .stats-section { grid-template-columns: 1fr; } diff --git a/src/commands/general/status.py b/src/commands/general/status.py index e470a6f..39cf536 100644 --- a/src/commands/general/status.py +++ b/src/commands/general/status.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import time from sqlalchemy import text @@ -22,19 +23,26 @@ class StatusCommand(Command): usage = "status" category = "general" - async def execute(self, ctx: CommandContext) -> None: - from commands.general.uptime import _start_time + def _ping_db(self) -> bool: + """Run blocking DB ping outside event loop.""" from core.db import get_engine - uptime = format_uptime(time.time() - _start_time) - - db_ok = False try: engine = get_engine() with engine.connect() as conn: conn.execute(text("SELECT 1")) - db_ok = True + return True except Exception: + return False + + async def execute(self, ctx: CommandContext) -> None: + from commands.general.uptime import _start_time + + uptime = format_uptime(time.time() - _start_time) + + try: + db_ok = await asyncio.wait_for(asyncio.to_thread(self._ping_db), timeout=2.0) + except TimeoutError: db_ok = False webhook = webhook_dispatcher_status() diff --git a/src/commands/owner/addskill.py b/src/commands/owner/addskill.py index 8e148ed..09c8324 100644 --- a/src/commands/owner/addskill.py +++ b/src/commands/owner/addskill.py @@ -44,6 +44,14 @@ def build_inline_skill(raw_args: str, quoted_text: str = "") -> dict[str, Any] | } +def decode_skill_document(media_data: bytes) -> str: + """Decode markdown skill document bytes. + + `utf-8-sig` handles both plain UTF-8 and UTF-8 BOM files. + """ + return media_data.decode("utf-8-sig") + + class AddSkillCommand(Command): name = "addskill" aliases = ["skill"] @@ -92,7 +100,7 @@ async def execute(self, ctx: CommandContext) -> None: try: media_data = await ctx.client._client.download_any(msg_obj) if media_data: - content = media_data.decode("utf-8") + content = decode_skill_document(media_data) skill = parse_skill_markdown(content) if skill: save_skill_to_file(skill) diff --git a/src/commands/utility/_ai_text.py b/src/commands/utility/_ai_text.py index 6d64b59..e982771 100644 --- a/src/commands/utility/_ai_text.py +++ b/src/commands/utility/_ai_text.py @@ -31,7 +31,12 @@ def ensure_provider_key(provider: str, api_key: str) -> None: apply_provider_env(provider, api_key) -async def ensure_ai_ready_or_reply(ctx: CommandContext, disabled_key: str) -> bool: +async def ensure_ai_ready_or_reply( + ctx: CommandContext, + disabled_key: str, + *, + no_key_key: str = "summarize.no_api_key", +) -> bool: """Validate AI enabled + API key; reply with localized error when invalid.""" ai_enabled = runtime_config.get_nested("agentic_ai", "enabled", default=False) if not ai_enabled: @@ -40,7 +45,7 @@ async def ensure_ai_ready_or_reply(ctx: CommandContext, disabled_key: str) -> bo api_key = get_api_key() if not api_key: - await ctx.client.reply(ctx.message, t_error("summarize.no_api_key")) + await ctx.client.reply(ctx.message, t_error(no_key_key)) return False return True diff --git a/src/commands/utility/rewrite.py b/src/commands/utility/rewrite.py index 892b059..2259e42 100644 --- a/src/commands/utility/rewrite.py +++ b/src/commands/utility/rewrite.py @@ -30,7 +30,11 @@ class RewriteCommand(Command): cooldown = 10 async def execute(self, ctx: CommandContext) -> None: - if not await ensure_ai_ready_or_reply(ctx, "rewrite.ai_disabled"): + if not await ensure_ai_ready_or_reply( + ctx, + "rewrite.ai_disabled", + no_key_key="rewrite.no_api_key", + ): return if not ctx.args: diff --git a/src/commands/utility/summarize.py b/src/commands/utility/summarize.py index 4b2cfab..cb2a0b5 100644 --- a/src/commands/utility/summarize.py +++ b/src/commands/utility/summarize.py @@ -30,7 +30,11 @@ class SummarizeCommand(Command): async def execute(self, ctx: CommandContext) -> None: """Summarize quoted text or recent chat memory.""" - if not await ensure_ai_ready_or_reply(ctx, "summarize.ai_disabled"): + if not await ensure_ai_ready_or_reply( + ctx, + "summarize.ai_disabled", + no_key_key="summarize.no_api_key", + ): return text_to_summarize = "" diff --git a/src/commands/utility/translate.py b/src/commands/utility/translate.py index 1d86fac..e5a4d87 100644 --- a/src/commands/utility/translate.py +++ b/src/commands/utility/translate.py @@ -22,7 +22,11 @@ class TranslateCommand(Command): cooldown = 10 async def execute(self, ctx: CommandContext) -> None: - if not await ensure_ai_ready_or_reply(ctx, "translate.ai_disabled"): + if not await ensure_ai_ready_or_reply( + ctx, + "translate.ai_disabled", + no_key_key="translate.no_api_key", + ): return if not ctx.args: diff --git a/src/locales/en.json b/src/locales/en.json index 8b4c9ac..f12d8eb 100644 --- a/src/locales/en.json +++ b/src/locales/en.json @@ -771,6 +771,7 @@ "title": "Translation", "processing": "Translating...", "ai_disabled": "AI is not enabled. Enable it with `config ai on`.", + "no_api_key": "AI API key is not configured.", "usage": "Usage: `{prefix}translate [text]` or reply to a message.", "no_text": "Provide text after the language or reply to a message.", "target": "Target language", @@ -780,6 +781,7 @@ "title": "Rewrite", "processing": "Rewriting...", "ai_disabled": "AI is not enabled. Enable it with `config ai on`.", + "no_api_key": "AI API key is not configured.", "usage": "Usage: `{prefix}rewrite [text]` or reply to a message.", "invalid_style": "Invalid style. Choose one of: {styles}", "no_text": "Provide text to rewrite or reply to a message.", diff --git a/src/locales/id.json b/src/locales/id.json index ee7c69c..f1a5eff 100644 --- a/src/locales/id.json +++ b/src/locales/id.json @@ -772,6 +772,7 @@ "title": "Terjemahan", "processing": "Lagi nerjemahin...", "ai_disabled": "AI belum diaktifin. Aktifin pake `config ai on`.", + "no_api_key": "API key AI belum di-set.", "usage": "Cara pake: `{prefix}translate [teks]` atau reply pesan.", "no_text": "Kasih teks setelah bahasa tujuan atau reply pesan.", "target": "Bahasa tujuan", @@ -781,6 +782,7 @@ "title": "Tulis Ulang", "processing": "Lagi nulis ulang...", "ai_disabled": "AI belum diaktifin. Aktifin pake `config ai on`.", + "no_api_key": "API key AI belum di-set.", "usage": "Cara pake: `{prefix}rewrite [teks]` atau reply pesan.", "invalid_style": "Style gak valid. Pilih salah satu: {styles}", "no_text": "Kasih teks yang mau ditulis ulang atau reply pesan.", diff --git a/tests/test_addskill.py b/tests/test_addskill.py deleted file mode 100644 index 4d3250e..0000000 --- a/tests/test_addskill.py +++ /dev/null @@ -1,20 +0,0 @@ -from commands.owner.addskill import build_inline_skill - - -def test_build_inline_skill_from_inline_text(): - skill = build_inline_skill("translator Always translate to English") - assert skill is not None - assert skill["name"] == "translator" - assert "Always translate" in skill["content"] - assert skill["trigger"] == "always" - - -def test_build_inline_skill_from_quoted_text(): - skill = build_inline_skill("translator", quoted_text="Translate to Indonesian") - assert skill is not None - assert skill["name"] == "translator" - assert skill["content"] == "Translate to Indonesian" - - -def test_build_inline_skill_requires_content(): - assert build_inline_skill("translator") is None diff --git a/tests/test_ai_action_policy.py b/tests/test_ai_action_policy.py deleted file mode 100644 index 7947d85..0000000 --- a/tests/test_ai_action_policy.py +++ /dev/null @@ -1,30 +0,0 @@ -from ai import agent as ai_agent - - -def test_blocked_actions_are_enforced(monkeypatch): - def fake_get_nested(*keys, default=None): - if keys == ("agentic_ai", "allowed_actions"): - return ["ping", "eval"] - if keys == ("agentic_ai", "blocked_actions"): - return ["eval"] - return default - - monkeypatch.setattr(ai_agent.runtime_config, "get_nested", fake_get_nested) - - assert ai_agent._is_ai_action_allowed("ping") - assert not ai_agent._is_ai_action_allowed("eval") - assert not ai_agent._is_ai_action_allowed("help") - - -def test_allowed_actions_empty_uses_blocklist_only(monkeypatch): - def fake_get_nested(*keys, default=None): - if keys == ("agentic_ai", "allowed_actions"): - return [] - if keys == ("agentic_ai", "blocked_actions"): - return ["shutdown"] - return default - - monkeypatch.setattr(ai_agent.runtime_config, "get_nested", fake_get_nested) - - assert ai_agent._is_ai_action_allowed("ping") - assert not ai_agent._is_ai_action_allowed("shutdown") diff --git a/tests/test_automations.py b/tests/test_automations.py deleted file mode 100644 index 8a2f2e1..0000000 --- a/tests/test_automations.py +++ /dev/null @@ -1,62 +0,0 @@ -from pathlib import Path - -import core.db as db_module -from core.automations import ( - get_automation_runtime, - rule_matches, - set_automation_dry_run, -) - - -def _reset_db(tmp_path: Path, monkeypatch) -> None: - db_file = tmp_path / "automations.db" - monkeypatch.setenv("DATABASE_URL", f"sqlite:///{db_file.as_posix()}") - db_module._engine = None - db_module._ready = False - db_module.ensure_database_ready() - - -def test_rule_matches_text_variants(): - assert rule_matches({"trigger_type": "contains", "trigger_value": "hello"}, "hello world") - assert rule_matches({"trigger_type": "starts_with", "trigger_value": "!promo"}, "!promo now") - assert rule_matches({"trigger_type": "exact_match", "trigger_value": "Hello"}, " hello ") - assert rule_matches({"trigger_type": "regex", "trigger_value": r"free\s+money"}, "FREE money") - assert rule_matches({"trigger_type": "link", "trigger_value": ""}, "visit https://example.com") - - assert not rule_matches( - {"trigger_type": "starts_with", "trigger_value": "!promo"}, - "check !promo", - ) - assert not rule_matches( - {"trigger_type": "exact_match", "trigger_value": "hello"}, - "hello world", - ) - - -def test_rule_matches_media_type(): - assert rule_matches( - {"trigger_type": "media_type", "trigger_value": "image"}, - "", - media_type="image", - ) - assert not rule_matches( - {"trigger_type": "media_type", "trigger_value": "image"}, - "", - media_type="video", - ) - - -def test_automation_dry_run_runtime_roundtrip(tmp_path, monkeypatch): - _reset_db(tmp_path, monkeypatch) - group = "12345@g.us" - - runtime = get_automation_runtime(group) - assert runtime["dry_run"] is False - - set_automation_dry_run(group, True) - runtime = get_automation_runtime(group) - assert runtime["dry_run"] is True - - set_automation_dry_run(group, False) - runtime = get_automation_runtime(group) - assert runtime["dry_run"] is False diff --git a/tests/test_command_permission_overrides.py b/tests/test_command_permission_overrides.py deleted file mode 100644 index 4749269..0000000 --- a/tests/test_command_permission_overrides.py +++ /dev/null @@ -1,142 +0,0 @@ -import pytest - -import core.permissions as permissions_module -from core.permissions import check_command_permissions -from core.types import ChatType - - -class DummyCommand: - def __init__( - self, - name: str, - *, - owner_only: bool = False, - admin_only: bool = False, - ): - self.name = name - self.owner_only = owner_only - self.admin_only = admin_only - self.bot_admin_required = False - self.group_only = False - self.private_only = False - - def can_execute(self, chat_type): - return chat_type in {ChatType.PRIVATE, ChatType.GROUP} - - -class DummyMessage: - def __init__( - self, - text: str, - *, - sender_jid: str = "111@s.whatsapp.net", - chat_jid: str = "123@g.us", - is_group: bool = True, - ): - self.text = text - self.sender_jid = sender_jid - self.chat_jid = chat_jid - self.is_group = is_group - self.chat_type = ChatType.GROUP if is_group else ChatType.PRIVATE - - -async def _owner_false(_sender_jid, _bot): - return False - - -@pytest.mark.asyncio -async def test_global_admin_override_blocks_member(monkeypatch): - async def _is_admin(_bot, _group_jid, _user_jid): - return False - - monkeypatch.setattr( - permissions_module.runtime_config, "get_owner_jid", lambda: "owner@s.whatsapp.net" - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _owner_false) - monkeypatch.setattr(permissions_module, "check_admin_permission", _is_admin) - monkeypatch.setattr( - permissions_module.runtime_config, - "get_command_role_override", - lambda name, group_jid=None: "admin" if name == "ping" else None, - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_command_enabled", lambda _name: True) - - cmd = DummyCommand("ping") - msg = DummyMessage("/ping", is_group=True) - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is False - - -@pytest.mark.asyncio -async def test_global_admin_override_allows_admin(monkeypatch): - async def _is_admin(_bot, _group_jid, _user_jid): - return True - - monkeypatch.setattr( - permissions_module.runtime_config, "get_owner_jid", lambda: "owner@s.whatsapp.net" - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _owner_false) - monkeypatch.setattr(permissions_module, "check_admin_permission", _is_admin) - monkeypatch.setattr( - permissions_module.runtime_config, - "get_command_role_override", - lambda name, group_jid=None: "admin" if name == "ping" else None, - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_command_enabled", lambda _name: True) - - cmd = DummyCommand("ping") - msg = DummyMessage("/ping", is_group=True) - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is True - - -@pytest.mark.asyncio -async def test_group_override_can_relax_admin_to_member(monkeypatch): - async def _is_admin(_bot, _group_jid, _user_jid): - return False - - monkeypatch.setattr( - permissions_module.runtime_config, "get_owner_jid", lambda: "owner@s.whatsapp.net" - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _owner_false) - monkeypatch.setattr(permissions_module, "check_admin_permission", _is_admin) - monkeypatch.setattr( - permissions_module.runtime_config, - "get_command_role_override", - lambda name, group_jid=None: "member" - if name == "warn" and group_jid == "123@g.us" - else None, - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_command_enabled", lambda _name: True) - - cmd = DummyCommand("warn", admin_only=True) - msg = DummyMessage("/warn @user") - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is True - - -@pytest.mark.asyncio -async def test_owner_only_cannot_be_relaxed(monkeypatch): - async def _is_admin(_bot, _group_jid, _user_jid): - return True - - monkeypatch.setattr( - permissions_module.runtime_config, "get_owner_jid", lambda: "owner@s.whatsapp.net" - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _owner_false) - monkeypatch.setattr(permissions_module, "check_admin_permission", _is_admin) - monkeypatch.setattr( - permissions_module.runtime_config, - "get_command_role_override", - lambda name, group_jid=None: "member" if name == "eval" else None, - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_command_enabled", lambda _name: True) - - cmd = DummyCommand("eval", owner_only=True) - msg = DummyMessage("/eval 1+1") - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is False diff --git a/tests/test_config_diff_image.py b/tests/test_config_diff_image.py deleted file mode 100644 index 14fd30d..0000000 --- a/tests/test_config_diff_image.py +++ /dev/null @@ -1,34 +0,0 @@ -from commands.owner.config import ( - _is_sensitive_path, - _mask_value, - _resolve_diff_mode, - render_diff_image, -) - - -def test_sensitive_path_detection_and_masking(): - assert _is_sensitive_path("agentic_ai.api_key") - assert _is_sensitive_path("dashboard.password") - assert not _is_sensitive_path("bot.prefix") - - assert _mask_value("agentic_ai.api_key", "sk-123") == "[redacted]" - assert _mask_value("bot.prefix", "!") == "!" - - -def test_render_diff_image_returns_png_bytes(): - rows = [ - ("~ bot.prefix: / -> !", "changed"), - ("+ privacy.ai_memory_enabled: True", "added"), - ("- features.old_flag: False", "missing"), - ] - data = render_diff_image("Config Differences", rows) - assert isinstance(data, bytes) - assert data.startswith(b"\x89PNG\r\n\x1a\n") - - -def test_resolve_diff_mode_defaults_image_and_supports_text(): - assert _resolve_diff_mode(None) == "image" - assert _resolve_diff_mode([]) == "image" - assert _resolve_diff_mode(["text"]) == "text" - assert _resolve_diff_mode(["txt"]) == "text" - assert _resolve_diff_mode(["image"]) == "image" diff --git a/tests/test_dashboard_security_and_ratelimit.py b/tests/test_dashboard_security_and_ratelimit.py deleted file mode 100644 index 691103e..0000000 --- a/tests/test_dashboard_security_and_ratelimit.py +++ /dev/null @@ -1,126 +0,0 @@ -import hashlib -import hmac -import time - -import pytest -from fastapi import HTTPException - -import dashboard_api - - -def test_dashboard_credentials_require_env(monkeypatch): - monkeypatch.delenv("DASHBOARD_USERNAME", raising=False) - monkeypatch.delenv("DASHBOARD_PASSWORD", raising=False) - - with pytest.raises(HTTPException) as exc: - dashboard_api._get_dashboard_credentials() - - assert exc.value.status_code == 503 - - -def test_dashboard_credentials_reject_admin_defaults(monkeypatch): - monkeypatch.setenv("DASHBOARD_USERNAME", "admin") - monkeypatch.setenv("DASHBOARD_PASSWORD", "admin") - - with pytest.raises(HTTPException) as exc: - dashboard_api._get_dashboard_credentials() - - assert exc.value.status_code == 503 - - -@pytest.mark.asyncio -async def test_rate_limit_update_persists(monkeypatch): - captured = {} - limiter = {} - emitted = [] - - def fake_set(key, value): - captured[key] = value - - def fake_update(config): - limiter.update(config.__dict__) - - async def fake_emit(event_type, payload): - emitted.append((event_type, payload)) - - monkeypatch.setattr(dashboard_api.runtime_config, "set", fake_set) - monkeypatch.setattr(dashboard_api.rate_limiter, "update_config", fake_update) - monkeypatch.setattr(dashboard_api.event_bus, "emit", fake_emit) - - result = await dashboard_api.update_rate_limit( - dashboard_api.RateLimitSettings( - enabled=True, - user_cooldown=4.5, - command_cooldown=3.0, - burst_limit=9, - burst_window=12.0, - ) - ) - - assert result == {"success": True} - assert captured["rate_limit"]["burst_limit"] == 9 - assert limiter["burst_limit"] == 9 - assert emitted[0][0] == "config_update" - - -def test_incoming_signature_verification(): - token = "test-token" - timestamp = str(int(time.time())) - payload = b'{"action":"emit_event","data":{}}' - digest = hmac.new( - token.encode("utf-8"), f"{timestamp}.".encode() + payload, hashlib.sha256 - ).hexdigest() - signature = f"sha256={digest}" - - assert dashboard_api._verify_incoming_signature(token, timestamp, signature, payload) - assert not dashboard_api._verify_incoming_signature(token, timestamp, "sha256=bad", payload) - - -def test_incoming_rate_limiter_window(): - key_id = 9999 - dashboard_api._incoming_rate_windows.pop(key_id, None) - - assert dashboard_api._consume_incoming_rate_limit(key_id, 2) - assert dashboard_api._consume_incoming_rate_limit(key_id, 2) - assert not dashboard_api._consume_incoming_rate_limit(key_id, 2) - - -class _FakeRequest: - def __init__(self, headers: dict[str, str], payload: bytes): - self.headers = headers - self._payload = payload - - async def body(self) -> bytes: - return self._payload - - -@pytest.mark.asyncio -async def test_incoming_webhook_duplicate_idempotency_returns_409(monkeypatch): - payload = b'{"action":"emit_event","data":{"event_type":"ci_done","event_data":{}}}' - request = _FakeRequest( - headers={ - "X-ZeroIchi-Incoming-Timestamp": str(int(time.time())), - "X-ZeroIchi-Incoming-Signature": "sha256=test", - "X-ZeroIchi-Incoming-Idempotency-Key": "dup-1", - }, - payload=payload, - ) - - monkeypatch.setattr( - dashboard_api, - "resolve_incoming_webhook_key", - lambda _token: { - "id": 1, - "enabled": True, - "allowed_actions": ["emit_event"], - "rate_limit_per_minute": 30, - }, - ) - monkeypatch.setattr(dashboard_api, "_verify_incoming_signature", lambda *_: True) - monkeypatch.setattr(dashboard_api, "_consume_incoming_rate_limit", lambda *_: True) - monkeypatch.setattr(dashboard_api, "claim_incoming_idempotency", lambda *_: False) - - with pytest.raises(HTTPException) as exc: - await dashboard_api.incoming_webhook_endpoint("token", request) # type: ignore[arg-type] - - assert exc.value.status_code == 409 diff --git a/tests/test_db_postgresql.py b/tests/test_db_postgresql.py deleted file mode 100644 index 69b0db0..0000000 --- a/tests/test_db_postgresql.py +++ /dev/null @@ -1,224 +0,0 @@ -""" -PostgreSQL integration tests — run against a live PostgreSQL database. - -These tests exercise all db.py operations against a real PostgreSQL instance -to verify dialect-specific branching (BIGSERIAL, RETURNING, etc.) works. - -Requires DATABASE_URL env var pointing to a PostgreSQL database. -Skipped automatically if DATABASE_URL is not set or not postgresql. -""" - -import os - -import pytest -from dotenv import load_dotenv - -load_dotenv() - -import core.db as db_module # noqa: E402 - -PG_URL = os.getenv("DATABASE_URL", "") -_is_pg = "postgresql" in PG_URL or "postgres" in PG_URL - -pytestmark = pytest.mark.skipif(not _is_pg, reason="DATABASE_URL not set to PostgreSQL") - - -@pytest.fixture(autouse=True) -def _pg_engine(): - """Reset db module state and initialize against the real PostgreSQL.""" - db_module._engine = None - db_module._ready = False - db_module.ensure_database_ready() - yield - engine = db_module.get_engine() - with engine.begin() as conn: - from sqlalchemy import text - - conn.execute(text("DELETE FROM webhook_deliveries")) - conn.execute(text("DELETE FROM webhooks")) - conn.execute(text("DELETE FROM incoming_webhook_keys")) - conn.execute(text("DELETE FROM kv_store WHERE scope LIKE 'test_%'")) - conn.execute(text("DELETE FROM audit_logs")) - - -def test_pg_connection(): - """Verify we are actually connected to PostgreSQL.""" - engine = db_module.get_engine() - assert engine.dialect.name == "postgresql" - - -def test_pg_kv_roundtrip(): - """KV store read/write on PostgreSQL.""" - payload = {"count": 42, "items": ["a", "b"]} - db_module.kv_set_json("test_pg", "stats", payload) - loaded = db_module.kv_get_json("test_pg", "stats", default={}) - assert loaded == payload - - db_module.kv_set_json("test_pg", "stats", {"count": 99}) - updated = db_module.kv_get_json("test_pg", "stats", default={}) - assert updated["count"] == 99 - - db_module.kv_delete("test_pg", "stats") - gone = db_module.kv_get_json("test_pg", "stats", default=None) - assert gone is None - - -def test_pg_webhook_crud(): - """Webhook CRUD with BIGSERIAL id and RETURNING on PostgreSQL.""" - hook = db_module.create_webhook( - name="PG Test", - url="https://example.com/pg-hook", - events=["command_executed", "message_received"], - secret="pg_secret", - enabled=True, - ) - - assert hook["id"] > 0 - assert hook["name"] == "PG Test" - assert hook["enabled"] is True - - fetched = db_module.get_webhook(hook["id"]) - assert fetched is not None - assert fetched["url"] == "https://example.com/pg-hook" - - updated = db_module.update_webhook(hook["id"], enabled=False) - assert updated is not None - assert updated["enabled"] is False - - matches = db_module.get_active_webhooks_for_event("command_executed") - assert all(m["id"] != hook["id"] for m in matches) - - db_module.update_webhook(hook["id"], enabled=True) - matches = db_module.get_active_webhooks_for_event("command_executed") - assert any(m["id"] == hook["id"] for m in matches) - - assert db_module.delete_webhook(hook["id"]) - assert db_module.get_webhook(hook["id"]) is None - - -def test_pg_webhook_delivery_log(): - """Webhook delivery logging with BIGINT foreign key on PostgreSQL.""" - hook = db_module.create_webhook( - name="Delivery Test", - url="https://example.com/deliver", - events=["*"], - secret="s", - enabled=True, - ) - - db_module.record_webhook_delivery( - webhook_id=hook["id"], - event_type="test_event", - payload={"ok": True}, - success=True, - attempt=1, - status_code=200, - ) - - deliveries = db_module.list_webhook_deliveries(hook["id"], limit=10) - assert len(deliveries) == 1 - assert deliveries[0]["success"] is True - assert deliveries[0]["status_code"] == 200 - - -def test_pg_webhook_auto_disable(): - """Auto-disable after max_failures on PostgreSQL.""" - hook = db_module.create_webhook( - name="Auto Disable PG", - url="https://example.com/fail", - events=["*"], - secret="s", - enabled=True, - max_failures=2, - ) - - db_module.mark_webhook_delivery_result(hook["id"], success=False, error="timeout") - h1 = db_module.get_webhook(hook["id"]) - assert h1["enabled"] is True - assert h1["failure_count"] == 1 - - db_module.mark_webhook_delivery_result(hook["id"], success=False, error="timeout") - h2 = db_module.get_webhook(hook["id"]) - assert h2["enabled"] is False - assert h2["failure_count"] == 2 - assert h2["disabled_reason"] is not None - - -def test_pg_incoming_webhook_key_crud(): - """Incoming webhook key CRUD with RETURNING on PostgreSQL.""" - created = db_module.create_incoming_webhook_key( - name="PG Incoming", - allowed_actions=["send_message", "emit_event"], - rate_limit_per_minute=30, - enabled=True, - ) - - assert created["id"] > 0 - assert created["token"] - - resolved = db_module.resolve_incoming_webhook_key(created["token"]) - assert resolved is not None - assert resolved["name"] == "PG Incoming" - assert resolved["rate_limit_per_minute"] == 30 - - new_token = db_module.rotate_incoming_webhook_key(created["id"]) - assert new_token - assert db_module.resolve_incoming_webhook_key(created["token"]) is None - assert db_module.resolve_incoming_webhook_key(new_token) is not None - - assert db_module.delete_incoming_webhook_key(created["id"]) - - -def test_pg_idempotency(): - """Idempotency claim deduplication on PostgreSQL.""" - key = db_module.create_incoming_webhook_key( - name="Idempotency PG", - allowed_actions=["emit_event"], - rate_limit_per_minute=10, - enabled=True, - ) - resolved = db_module.resolve_incoming_webhook_key(key["token"]) - assert resolved is not None - - first = db_module.claim_incoming_idempotency(int(resolved["id"]), "pg-dedup-123") - second = db_module.claim_incoming_idempotency(int(resolved["id"]), "pg-dedup-123") - - assert first is True - assert second is False - - third = db_module.claim_incoming_idempotency(int(resolved["id"]), "pg-dedup-456") - assert third is True - - -def test_pg_audit_log(): - """Audit log write/read on PostgreSQL.""" - db_module.add_audit_log( - action="test_action", - actor="test_user@s.whatsapp.net", - resource="test_resource", - details={"note": "PG audit test"}, - ) - - logs = db_module.list_audit_logs(limit=5) - assert len(logs) >= 1 - latest = logs[0] - assert latest["action"] == "test_action" - assert latest["actor"] == "test_user@s.whatsapp.net" - - -def test_pg_secret_rotation(): - """Webhook secret rotation on PostgreSQL.""" - hook = db_module.create_webhook( - name="Rotate PG", - url="https://example.com/rotate", - events=["*"], - secret="old_secret", - enabled=True, - ) - - new_secret = db_module.rotate_webhook_secret(hook["id"]) - assert new_secret - assert new_secret != "old_secret" - - fetched = db_module.get_webhook(hook["id"]) - assert fetched["secret"] == new_secret diff --git a/tests/test_db_webhooks.py b/tests/test_db_webhooks.py deleted file mode 100644 index 8f815da..0000000 --- a/tests/test_db_webhooks.py +++ /dev/null @@ -1,128 +0,0 @@ -from pathlib import Path - -import core.db as db_module - - -def _reset_db(tmp_path: Path, monkeypatch) -> None: - db_file = tmp_path / "test.db" - monkeypatch.setenv("DATABASE_URL", f"sqlite:///{db_file.as_posix()}") - db_module._engine = None - db_module._ready = False - db_module.ensure_database_ready() - - -def test_kv_roundtrip(tmp_path, monkeypatch): - _reset_db(tmp_path, monkeypatch) - - payload = {"count": 42, "items": ["a", "b"]} - db_module.kv_set_json("global", "stats", payload) - - loaded = db_module.kv_get_json("global", "stats", default={}) - assert loaded == payload - - -def test_webhook_crud_and_delivery_log(tmp_path, monkeypatch): - _reset_db(tmp_path, monkeypatch) - - hook = db_module.create_webhook( - name="CI", - url="https://example.com/hook", - events=["command_executed"], - secret="abc", - enabled=True, - ) - - assert hook["name"] == "CI" - assert hook["enabled"] is True - - matches = db_module.get_active_webhooks_for_event("command_executed") - assert len(matches) == 1 - assert matches[0]["id"] == hook["id"] - - db_module.record_webhook_delivery( - webhook_id=hook["id"], - event_type="command_executed", - payload={"ok": True}, - success=True, - attempt=1, - status_code=204, - ) - - deliveries = db_module.list_webhook_deliveries(hook["id"], limit=10) - assert len(deliveries) == 1 - assert deliveries[0]["success"] is True - assert deliveries[0]["status_code"] == 204 - - assert db_module.delete_webhook(hook["id"]) - assert db_module.get_webhook(hook["id"]) is None - - -def test_webhook_auto_disable_after_failures(tmp_path, monkeypatch): - _reset_db(tmp_path, monkeypatch) - - hook = db_module.create_webhook( - name="Auto Disable", - url="https://example.com/hook", - events=["*"], - secret="abc", - enabled=True, - max_failures=2, - ) - - db_module.mark_webhook_delivery_result(hook["id"], success=False, error="timeout") - first = db_module.get_webhook(hook["id"]) - assert first is not None - assert first["enabled"] is True - assert first["failure_count"] == 1 - - db_module.mark_webhook_delivery_result(hook["id"], success=False, error="timeout") - second = db_module.get_webhook(hook["id"]) - assert second is not None - assert second["enabled"] is False - assert second["failure_count"] == 2 - assert second["disabled_reason"] is not None - - -def test_incoming_webhook_key_crud(tmp_path, monkeypatch): - _reset_db(tmp_path, monkeypatch) - - created = db_module.create_incoming_webhook_key( - name="CI Trigger", - allowed_actions=["send_message", "emit_event"], - rate_limit_per_minute=25, - enabled=True, - ) - - assert created["id"] > 0 - assert created["token"] - - resolved = db_module.resolve_incoming_webhook_key(created["token"]) - assert resolved is not None - assert resolved["name"] == "CI Trigger" - assert resolved["rate_limit_per_minute"] == 25 - - rotated = db_module.rotate_incoming_webhook_key(created["id"]) - assert rotated - assert db_module.resolve_incoming_webhook_key(created["token"]) is None - assert db_module.resolve_incoming_webhook_key(rotated) is not None - - assert db_module.delete_incoming_webhook_key(created["id"]) - - -def test_claim_incoming_idempotency(tmp_path, monkeypatch): - _reset_db(tmp_path, monkeypatch) - - key = db_module.create_incoming_webhook_key( - name="CI Trigger", - allowed_actions=["emit_event"], - rate_limit_per_minute=10, - enabled=True, - ) - resolved = db_module.resolve_incoming_webhook_key(key["token"]) - assert resolved is not None - - first = db_module.claim_incoming_idempotency(int(resolved["id"]), "abc-123") - second = db_module.claim_incoming_idempotency(int(resolved["id"]), "abc-123") - - assert first is True - assert second is False diff --git a/tests/test_id_utils.py b/tests/test_id_utils.py deleted file mode 100644 index 4d2b2b6..0000000 --- a/tests/test_id_utils.py +++ /dev/null @@ -1,11 +0,0 @@ -from core.id_utils import next_prefixed_id - - -def test_next_prefixed_id_basic_sequence(): - rows = [{"id": "A001"}, {"id": "A003"}, {"id": "A002"}] - assert next_prefixed_id(rows, prefix="A", width=3) == "A004" - - -def test_next_prefixed_id_ignores_invalid_values(): - rows = [{"id": "X999"}, {"id": "AAB"}, {"id": ""}, {}] - assert next_prefixed_id(rows, prefix="A", width=3) == "A001" diff --git a/tests/test_message_constants.py b/tests/test_message_constants.py deleted file mode 100644 index 849aa3c..0000000 --- a/tests/test_message_constants.py +++ /dev/null @@ -1,5 +0,0 @@ -from core.constants import TEXT_SOURCES - - -def test_document_caption_is_in_text_sources(): - assert ("documentMessage", "caption") in TEXT_SOURCES diff --git a/tests/test_owner_bootstrap_permissions.py b/tests/test_owner_bootstrap_permissions.py deleted file mode 100644 index 3364118..0000000 --- a/tests/test_owner_bootstrap_permissions.py +++ /dev/null @@ -1,94 +0,0 @@ -import pytest - -import core.permissions as permissions_module -from core.permissions import check_command_permissions -from core.types import ChatType - - -class DummyCommand: - def __init__(self, name: str, owner_only: bool = True): - self.name = name - self.owner_only = owner_only - self.admin_only = False - self.bot_admin_required = False - self.group_only = False - self.private_only = False - - def can_execute(self, chat_type): - return chat_type in {ChatType.PRIVATE, ChatType.GROUP} - - -class DummyMessage: - def __init__(self, text: str, is_group: bool = False): - self.text = text - self.is_group = is_group - self.chat_type = ChatType.GROUP if is_group else ChatType.PRIVATE - self.sender_jid = "12345@s.whatsapp.net" - - -async def _false_owner(_sender_jid, _bot): - return False - - -@pytest.mark.asyncio -async def test_config_owner_me_allowed_when_owner_not_set(monkeypatch): - monkeypatch.setattr(permissions_module.runtime_config, "get_owner_jid", lambda: "") - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _false_owner) - - cmd = DummyCommand("config", owner_only=True) - msg = DummyMessage("/config owner me") - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is True - - -@pytest.mark.asyncio -async def test_config_all_blocked_when_owner_not_set(monkeypatch): - monkeypatch.setattr(permissions_module.runtime_config, "get_owner_jid", lambda: "") - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _false_owner) - - cmd = DummyCommand("config", owner_only=True) - msg = DummyMessage("/config all") - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is False - - -@pytest.mark.asyncio -async def test_setup_start_allowed_when_owner_not_set(monkeypatch): - monkeypatch.setattr(permissions_module.runtime_config, "get_owner_jid", lambda: "") - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _false_owner) - - cmd = DummyCommand("setup", owner_only=True) - msg = DummyMessage("/setup start") - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is True - - -@pytest.mark.asyncio -async def test_setup_write_blocked_when_owner_not_set(monkeypatch): - monkeypatch.setattr(permissions_module.runtime_config, "get_owner_jid", lambda: "") - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _false_owner) - - cmd = DummyCommand("setup", owner_only=True) - msg = DummyMessage("/setup anti-link on warn") - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is False - - -@pytest.mark.asyncio -async def test_owner_only_still_blocked_when_owner_set(monkeypatch): - monkeypatch.setattr( - permissions_module.runtime_config, - "get_owner_jid", - lambda: "owner@s.whatsapp.net", - ) - monkeypatch.setattr(permissions_module.runtime_config, "is_owner_async", _false_owner) - - cmd = DummyCommand("config", owner_only=True) - msg = DummyMessage("/config owner me") - - result = await check_command_permissions(cmd, msg, bot=object()) - assert result.allowed is False diff --git a/tests/test_presentation.py b/tests/test_presentation.py deleted file mode 100644 index 5964b7f..0000000 --- a/tests/test_presentation.py +++ /dev/null @@ -1,20 +0,0 @@ -from core.presentation import format_command_card - - -def test_format_command_card_contains_expected_sections(): - text = format_command_card( - ".", - "permission", - "Manage role overrides for command access", - ".permission list | set | reset", - aliases=["permissions", "perm"], - category="owner", - restrictions=["Owner only"], - ) - - assert "「 `.permission` 」" in text - assert "» Manage role overrides for command access" in text - assert "• *Usage:* `.permission list | set | reset`" in text - assert "• *Aliases:* `.permissions`, `.perm`" in text - assert "• *Category:* ⛯ Owner" in text - assert "⊘ *Restrictions:* Owner only" in text diff --git a/tests/test_privacy_controls.py b/tests/test_privacy_controls.py deleted file mode 100644 index 897b2d0..0000000 --- a/tests/test_privacy_controls.py +++ /dev/null @@ -1,68 +0,0 @@ -from pathlib import Path - -import ai.memory as memory_module -import core.db as db_module -from ai.memory import clear_memory, get_memory -from core.privacy import ( - clear_chat_memory_override, - get_analytics_retention_days, - get_chat_memory_override, - is_chat_memory_enabled, - set_chat_memory_enabled, -) - - -def _reset_db(tmp_path: Path, monkeypatch) -> None: - db_file = tmp_path / "privacy_controls.db" - monkeypatch.setenv("DATABASE_URL", f"sqlite:///{db_file.as_posix()}") - db_module._engine = None - db_module._ready = False - db_module.ensure_database_ready() - - -def test_chat_memory_override_roundtrip(tmp_path, monkeypatch): - _reset_db(tmp_path, monkeypatch) - - chat = "123@g.us" - assert get_chat_memory_override(chat) is None - - set_chat_memory_enabled(chat, False) - assert get_chat_memory_override(chat) is False - assert is_chat_memory_enabled(chat) is False - - set_chat_memory_enabled(chat, True) - assert get_chat_memory_override(chat) is True - assert is_chat_memory_enabled(chat) is True - - assert clear_chat_memory_override(chat) is True - assert get_chat_memory_override(chat) is None - - -def test_analytics_retention_days_clamped(monkeypatch): - monkeypatch.setattr( - "core.privacy.runtime_config.get_nested", - lambda *_args, **_kwargs: 999, - ) - assert get_analytics_retention_days() == 365 - - monkeypatch.setattr( - "core.privacy.runtime_config.get_nested", - lambda *_args, **_kwargs: -5, - ) - assert get_analytics_retention_days() == 1 - - -def test_clear_memory_for_uncached_chat(tmp_path, monkeypatch): - _reset_db(tmp_path, monkeypatch) - - chat = "987@g.us" - mem = get_memory(chat, ttl_hours=24) - mem.add(role="user", content="hello") - assert len(mem.get_history()) == 1 - - memory_module._memory_cache.pop(chat, None) - - clear_memory(chat) - - reloaded = get_memory(chat, ttl_hours=24) - assert len(reloaded.get_history()) == 0 diff --git a/tests/test_runtime_config_history.py b/tests/test_runtime_config_history.py deleted file mode 100644 index 0d48884..0000000 --- a/tests/test_runtime_config_history.py +++ /dev/null @@ -1,59 +0,0 @@ -from pathlib import Path - -import pytest - -import core.runtime_config as runtime_config_module - - -@pytest.fixture -def isolated_runtime_config(tmp_path, monkeypatch): - schema_path = Path(__file__).resolve().parents[1] / "config.schema.json" - - monkeypatch.setattr(runtime_config_module, "CONFIG_FILE", tmp_path / "config.json") - monkeypatch.setattr(runtime_config_module, "SCHEMA_FILE", schema_path) - monkeypatch.setattr( - runtime_config_module, "OVERRIDES_FILE", tmp_path / "runtime_overrides.json" - ) - monkeypatch.setattr( - runtime_config_module, - "OVERRIDES_MIGRATION_MARKER", - tmp_path / ".runtime_overrides_migrated", - ) - monkeypatch.setattr(runtime_config_module, "HISTORY_FILE", tmp_path / "config_history.json") - runtime_config_module.RuntimeConfig._instance = None - - cfg = runtime_config_module.RuntimeConfig() - yield cfg - - runtime_config_module.RuntimeConfig._instance = None - - -def test_history_records_on_config_updates(isolated_runtime_config): - cfg = isolated_runtime_config - - cfg.set_nested("bot", "prefix", "!") - - entries = cfg.list_config_history(limit=5) - assert entries - assert entries[0]["id"].startswith("H") - assert entries[0]["reason"] == "update" - - -def test_rollback_restores_snapshot_config(isolated_runtime_config): - cfg = isolated_runtime_config - - cfg.set_nested("bot", "prefix", "!") - cfg.set_nested("bot", "prefix", "#") - assert cfg.get_nested("bot", "prefix") == "#" - - result = cfg.rollback_config("H0001") - assert result is not None - assert cfg.get_nested("bot", "prefix") == "/" - - -def test_rollback_unknown_id_returns_none(isolated_runtime_config): - cfg = isolated_runtime_config - - cfg.set_nested("bot", "prefix", "!") - result = cfg.rollback_config("H9999") - assert result is None diff --git a/tests/test_runtime_config_validation.py b/tests/test_runtime_config_validation.py deleted file mode 100644 index 4a026cd..0000000 --- a/tests/test_runtime_config_validation.py +++ /dev/null @@ -1,201 +0,0 @@ -from pathlib import Path - -import pytest - -import core.runtime_config as runtime_config_module - - -@pytest.fixture -def isolated_runtime_config(tmp_path, monkeypatch): - schema_path = Path(__file__).resolve().parents[1] / "config.schema.json" - - monkeypatch.setattr(runtime_config_module, "CONFIG_FILE", tmp_path / "config.json") - monkeypatch.setattr(runtime_config_module, "SCHEMA_FILE", schema_path) - monkeypatch.setattr( - runtime_config_module, "OVERRIDES_FILE", tmp_path / "runtime_overrides.json" - ) - monkeypatch.setattr( - runtime_config_module, - "OVERRIDES_MIGRATION_MARKER", - tmp_path / ".runtime_overrides_migrated", - ) - monkeypatch.setattr(runtime_config_module, "HISTORY_FILE", tmp_path / "config_history.json") - runtime_config_module.RuntimeConfig._instance = None - - cfg = runtime_config_module.RuntimeConfig() - yield cfg - - runtime_config_module.RuntimeConfig._instance = None - - -def test_invalid_schema_update_is_rejected(isolated_runtime_config): - cfg = isolated_runtime_config - - before = cfg.get_nested("rate_limit", "burst_limit") - - with pytest.raises(ValueError): - cfg.set_nested("rate_limit", "burst_limit", 0) - - assert cfg.get_nested("rate_limit", "burst_limit") == before - - -def test_valid_schema_update_is_persisted(isolated_runtime_config): - cfg = isolated_runtime_config - - cfg.set_nested("rate_limit", "burst_limit", 9) - - assert cfg.get_nested("rate_limit", "burst_limit") == 9 - - -def test_missing_schema_is_merged_and_preserved(tmp_path, monkeypatch): - schema_path = Path(__file__).resolve().parents[1] / "config.schema.json" - config_path = tmp_path / "config.json" - config_path.write_text( - """ -{ - "bot": { - "name": "Custom Bot", - "owner_jid": "12345@s.whatsapp.net" - }, - "features": { - "notes": false - } -} -""".strip(), - encoding="utf-8", - ) - - monkeypatch.setattr(runtime_config_module, "CONFIG_FILE", config_path) - monkeypatch.setattr(runtime_config_module, "SCHEMA_FILE", schema_path) - monkeypatch.setattr( - runtime_config_module, "OVERRIDES_FILE", tmp_path / "runtime_overrides.json" - ) - monkeypatch.setattr( - runtime_config_module, - "OVERRIDES_MIGRATION_MARKER", - tmp_path / ".runtime_overrides_migrated", - ) - monkeypatch.setattr(runtime_config_module, "HISTORY_FILE", tmp_path / "config_history.json") - runtime_config_module.RuntimeConfig._instance = None - - cfg = runtime_config_module.RuntimeConfig() - - assert cfg.get_nested("bot", "name") == "Custom Bot" - assert cfg.get_nested("bot", "owner_jid") == "12345@s.whatsapp.net" - assert cfg.get_nested("features", "notes") is False - assert cfg.get_nested("features", "anti_delete") is True - - persisted = runtime_config_module.jsonc.load(config_path) - assert persisted.get("$schema") == runtime_config_module.DEFAULT_SCHEMA_PATH - assert persisted.get("bot", {}).get("name") == "Custom Bot" - assert persisted.get("features", {}).get("notes") is False - - runtime_config_module.RuntimeConfig._instance = None - - -def test_invalid_config_does_not_overwrite_file(tmp_path, monkeypatch): - schema_path = Path(__file__).resolve().parents[1] / "config.schema.json" - config_path = tmp_path / "config.json" - invalid_content = """ -{ - "bot": "not-an-object" -} -""".strip() - config_path.write_text(invalid_content, encoding="utf-8") - - monkeypatch.setattr(runtime_config_module, "CONFIG_FILE", config_path) - monkeypatch.setattr(runtime_config_module, "SCHEMA_FILE", schema_path) - monkeypatch.setattr( - runtime_config_module, "OVERRIDES_FILE", tmp_path / "runtime_overrides.json" - ) - monkeypatch.setattr( - runtime_config_module, - "OVERRIDES_MIGRATION_MARKER", - tmp_path / ".runtime_overrides_migrated", - ) - monkeypatch.setattr(runtime_config_module, "HISTORY_FILE", tmp_path / "config_history.json") - runtime_config_module.RuntimeConfig._instance = None - - cfg = runtime_config_module.RuntimeConfig() - - assert cfg.get_nested("bot", "name") == "Zero Ichi" - assert config_path.read_text(encoding="utf-8").strip() == invalid_content - - runtime_config_module.RuntimeConfig._instance = None - - -def test_missing_default_keys_are_persisted_with_existing_schema(tmp_path, monkeypatch): - schema_path = Path(__file__).resolve().parents[1] / "config.schema.json" - config_path = tmp_path / "config.json" - config_path.write_text( - """ -{ - "$schema": "./config.schema.json", - "bot": { - "name": "Custom Bot", - "prefix": "/", - "login_method": "QR", - "phone_number": "", - "owner_jid": "owner@s.whatsapp.net", - "auto_read": false, - "auto_reload": true, - "auto_react": false, - "auto_react_emoji": "", - "ignore_self_messages": true, - "self_mode": false - }, - "features": { - "anti_delete": true, - "anti_link": true, - "welcome": true, - "notes": false, - "filters": true, - "blacklist": true, - "warnings": true, - "automation_rules": true - } -} -""".strip(), - encoding="utf-8", - ) - - monkeypatch.setattr(runtime_config_module, "CONFIG_FILE", config_path) - monkeypatch.setattr(runtime_config_module, "SCHEMA_FILE", schema_path) - monkeypatch.setattr( - runtime_config_module, "OVERRIDES_FILE", tmp_path / "runtime_overrides.json" - ) - monkeypatch.setattr( - runtime_config_module, - "OVERRIDES_MIGRATION_MARKER", - tmp_path / ".runtime_overrides_migrated", - ) - monkeypatch.setattr(runtime_config_module, "HISTORY_FILE", tmp_path / "config_history.json") - runtime_config_module.RuntimeConfig._instance = None - - cfg = runtime_config_module.RuntimeConfig() - - # Existing user values are preserved - assert cfg.get_nested("bot", "name") == "Custom Bot" - assert cfg.get_nested("features", "notes") is False - - # Newly added keys are merged and persisted - assert cfg.get_nested("features", "anti_spam") is False - persisted = runtime_config_module.jsonc.load(config_path) - assert persisted.get("features", {}).get("anti_spam") is False - - runtime_config_module.RuntimeConfig._instance = None - - -def test_command_role_override_roundtrip(isolated_runtime_config): - cfg = isolated_runtime_config - - assert cfg.get_command_role_override("warn") is None - cfg.set_command_role_override("warn", "admin") - assert cfg.get_command_role_override("warn") == "admin" - - cfg.set_command_role_override("quote", "owner", group_jid="123@g.us") - assert cfg.get_command_role_override("quote", group_jid="123@g.us") == "owner" - - assert cfg.reset_command_role_override("warn") is True - assert cfg.get_command_role_override("warn") is None - assert cfg.reset_command_role_override("warn") is False diff --git a/tests/test_timefmt.py b/tests/test_timefmt.py deleted file mode 100644 index 6ebecd3..0000000 --- a/tests/test_timefmt.py +++ /dev/null @@ -1,11 +0,0 @@ -from core.timefmt import format_uptime - - -def test_format_uptime_without_seconds(): - assert format_uptime(18 * 60) == "18m" - assert format_uptime(2 * 3600 + 5 * 60) == "2h 5m" - - -def test_format_uptime_with_seconds(): - assert format_uptime(59, include_seconds=True) == "59s" - assert format_uptime(61, include_seconds=True) == "1m 1s"