tests/test_security.py

"""test_security.py — Regression test sicurezza P19-SEC2 (audit 2026-07-08, rev 2026-07-09).

Esegui con: python3 -m pytest backend/tests/test_security.py -v
"""
from __future__ import annotations
import asyncio, os, sys, time, unittest
import unittest.mock
from datetime import datetime, timezone, timedelta
from fastapi import HTTPException

_B = os.path.join(os.path.dirname(__file__), "..")
if _B not in sys.path:
    sys.path.insert(0, _B)

# Helper per eseguire coroutine
def _run(coro):
    return asyncio.run(coro)

class TestShellValidate(unittest.TestCase):
    """SEC-1: validate_shell_command blocca tutti i vettori di injection."""

    def setUp(self):
        try:
            from tools._shell_safety import validate_shell_command, run_shell_safe
            self.v = validate_shell_command
            self.rs = run_shell_safe
        except ImportError as e:
            self.skipTest(str(e))

    # Comandi validi
    def test_ok_ls(self):         self.assertIsNone(self.v("ls"))
    def test_ok_ls_args(self):    self.assertIsNone(self.v("ls -la"))
    def test_ok_git_status(self): self.assertIsNone(self.v("git status"))
    def test_ok_curl_https(self): self.assertIsNone(self.v("curl https://example.com"))
    def test_ok_echo(self):       self.assertIsNone(self.v("echo hello"))

    # Metacaratteri shell (GAP-11 / GAP-12)
    def test_block_semicolon(self):
        self.assertIsNotNone(self.v("ls; rm -rf /"), "semicolon bypass non rilevato")

    def test_block_ampersand(self):
        self.assertIsNotNone(self.v("echo ok && curl x.com"), "ampersand bypass")

    def test_block_pipe(self):
        self.assertIsNotNone(self.v("ls | curl x.com"), "pipe bypass")

    def test_block_dollar_paren(self):
        self.assertIsNotNone(self.v("echo $(rm /)"), "$(cmd) bypass")

    def test_block_newline(self):
        self.assertIsNotNone(self.v("ls\nrm /"), "newline bypass")

    # Comandi non in allowlist
    def test_block_rm(self):
        self.assertIsNotNone(self.v("rm -rf /tmp"), "rm non bloccato")

    def test_block_git_push(self):
        self.assertIsNotNone(self.v("git push origin main"), "git push non bloccato")

    def test_block_curl_http(self):
        self.assertIsNotNone(self.v("curl http://attacker.com"), "curl http non bloccato")

    # run_shell_safe
    def test_run_blocks_injection(self):
        r = _run(self.rs("ls; whoami"))
        self.assertFalse(r["ok"])
        self.assertIsNotNone(r["error"])
        self.assertEqual(r["stdout"], "")

    def test_run_executes_valid(self):
        r = _run(self.rs("echo sec_test_ok"))
        self.assertTrue(r["ok"])
        self.assertIn("sec_test_ok", r["stdout"])


class TestSafeEnv(unittest.TestCase):
    """SEC-2: safe_shell_env non espone secret del processo padre."""

    def setUp(self):
        try:
            from tools._shell_safety import safe_shell_env
            self.fn = safe_shell_env
        except ImportError as e:
            self.skipTest(str(e))

    def _get(self, **kw):
        with unittest.mock.patch.dict(os.environ, kw):
            return self.fn()

    def test_no_internal_token(self):
        self.assertNotIn("INTERNAL_TOKEN", self._get(INTERNAL_TOKEN="secret"))

    def test_no_supabase_key(self):
        self.assertNotIn("SUPABASE_KEY", self._get(SUPABASE_KEY="srv_role"))

    def test_no_hf_token_a(self):
        self.assertNotIn("HF_TOKEN_A", self._get(HF_TOKEN_A="hf_fake"))

    def test_no_vault_key(self):
        self.assertNotIn("VAULT_KEY", self._get(VAULT_KEY="fernet_key"))


class TestRequireRole(unittest.TestCase):
    """SEC-3: require_role(MACHINE) non fa fail-open con INTERNAL_TOKEN vuoto."""

    def setUp(self):
        try:
            from api.auth_guard import require_role, AuthRole
            self.rr = require_role
            self.AR = AuthRole
        except ImportError as e:
            self.skipTest(str(e))

    def test_machine_rejects_empty_token(self):
        dep = self.rr(self.AR.MACHINE)
        req_mock = unittest.mock.MagicMock()
        req_mock.headers = {}
        with unittest.mock.patch.dict(os.environ, {"INTERNAL_TOKEN": ""}):
            with self.assertRaises(HTTPException) as ctx:
                # Passiamo resolved=AuthRole.USER per simulare la dependency non risolta
                # o testare il comportamento di require_role quando chiamata direttamente
                _run(dep(req_mock, resolved=self.AR.USER))
            self.assertIn(ctx.exception.status_code, (401, 403, 503),
                "require_role fail-open con INTERNAL_TOKEN vuoto")

    def test_machine_rejects_wrong_token(self):
        dep = self.rr(self.AR.MACHINE)
        req_mock = unittest.mock.MagicMock()
        req_mock.headers = {"authorization": "Bearer wrongtoken"}
        with unittest.mock.patch.dict(os.environ, {"INTERNAL_TOKEN": "correcttoken"}):
            with self.assertRaises(HTTPException):
                _run(dep(req_mock, resolved=self.AR.USER))


class TestMachineRateLimitClientBuckets(unittest.TestCase):
    """SEC-3B: il token proxy MACHINE non deve condividere un unico bucket pubblico."""

    def setUp(self):
        try:
            from api.auth_guard import _rate_key
            self.rate_key = _rate_key
        except ImportError as e:
            self.skipTest(str(e))

    def test_machine_uses_distinct_attested_client_ips(self):
        first = self.rate_key(1, "shared-internal-token", "198.51.100.10")
        second = self.rate_key(1, "shared-internal-token", "198.51.100.11")
        self.assertNotEqual(first, second)

    def test_machine_without_attested_ip_keeps_token_bucket(self):
        first = self.rate_key(1, "shared-internal-token", None)
        second = self.rate_key(1, "shared-internal-token", None)
        self.assertEqual(first, second)

    def test_operator_remains_token_scoped(self):
        first = self.rate_key(2, "operator-token", "198.51.100.10")
        second = self.rate_key(2, "operator-token", "198.51.100.11")
        self.assertEqual(first, second)


class TestBrowserEndpointAuth(unittest.TestCase):
    """SEC-5: endpoint browser stateless non fail-open (GAP-1-fix)."""

    def _check_role_param(self, fn_name: str):
        try:
            import api.browser as bmod
            fn = getattr(bmod, fn_name)
            import inspect
            sig = inspect.signature(fn)
            self.assertIn('role', sig.parameters,
                f"{fn_name} non ha parametro 'role' — require_role mancante")
        except (ImportError, AttributeError) as e:
            self.skipTest(str(e))

    def test_screenshot_auth(self):     self._check_role_param('browser_screenshot')
    def test_navigate_auth(self):       self._check_role_param('browser_navigate')
    def test_close_auth(self):          self._check_role_param('browser_close')
    def test_session_shot_auth(self):   self._check_role_param('browser_session_screenshot')
    def test_list_sessions_auth(self):  self._check_role_param('list_sessions')


class TestJsBlocklistAppliedToTS(unittest.TestCase):
    """SEC-6: _JS_BLOCKED_RE applicata anche a TypeScript (GAP-3-fix)."""

    def setUp(self):
        try:
            import api.exec as emod
            self._re = emod._JS_BLOCKED_RE
            import inspect
            self._exec_src = inspect.getsource(emod.exec_code)
        except (ImportError, AttributeError) as e:
            self.skipTest(str(e))

    def test_ts_blocked_child_process(self):
        """Il codice TS con require('child_process') deve essere bloccato."""
        dangerous = "import { exec } from 'child_process'; exec('whoami')"
        self.assertIsNotNone(self._re.search(dangerous), "child_process non bloccato in TS")

    def test_ts_block_check_in_source(self):
        """Il fix GAP-3 deve apparire nel sorgente di exec_code."""
        # Il codice usa 'js_blocked' come prefisso per entrambi JS e TS
        self.assertIn('js_blocked', self._exec_src, "GAP-3-fix non trovato in exec_code — TS check mancante")


# ─────────────────────────────────────────────────────────────────────────────
# SEC-7 → SEC-11  Audit 2026-07-09 — regression comportamentale GAP-4/6/7/8/9
# rev 2: SEC-8/10/11 riscritti da string-matching a test comportamentali reali
# dopo code review che segnalava falsi negativi possibili (vedi commit history).
# ─────────────────────────────────────────────────────────────────────────────

class TestUserIdFixed(unittest.TestCase):
    """SEC-7: _user_id() ritorna sempre 'default' — ignora X-User-ID client (GAP-4-fix)."""

    def setUp(self):
        try:
            from api.auth_managed import _user_id
            self._fn = _user_id
        except ImportError as e:
            self.skipTest(str(e))

    def _make_req(self, headers: dict):
        try:
            from starlette.requests import Request as _Req
        except ImportError as e:
            self.skipTest(str(e))
        scope = {
            "type": "http",
            "method": "GET",
            "path": "/test",
            "query_string": b"",
            "headers": [(k.lower().encode(), v.encode()) for k, v in headers.items()],
        }
        return _Req(scope)

    def test_returns_default_no_header(self):
        req = self._make_req({})
        self.assertEqual(self._fn(req), "default")

    def test_ignores_x_user_id_header(self):
        """GAP-4: X-User-ID dal client NON deve influenzare il risultato."""
        req = self._make_req({"X-User-ID": "attacker"})
        self.assertEqual(self._fn(req), "default",
            "REGRESSO GAP-4: _user_id ha letto X-User-ID — hijack possibile")

    def test_ignores_spoofed_admin(self):
        req = self._make_req({"X-User-ID": "admin", "x-forwarded-user": "root"})
        self.assertEqual(self._fn(req), "default")

    def test_ignores_multiple_hijack_headers_simultaneously(self):
        req = self._make_req({
            "X-User-ID": "admin",
            "X-Forwarded-User": "root",
            "X-Auth-User": "superuser",
        })
        self.assertEqual(self._fn(req), "default")


class TestCORSOriginValidationBehavior(unittest.TestCase):
    """SEC-8: _is_allowed_origin() valida realmente l'origin contro la whitelist (GAP-8-fix).

    Esegue il vero codice sorgente di main.py (blocco estratto, non riscritto) in un
    namespace isolato — evita gli import pesanti dell'intera app FastAPI (router,
    scheduler, DB) mantenendo comportamento reale, non string-matching.
    """

    def _load_fn(self, extra_env: dict | None = None):
        _mp = os.path.join(os.path.dirname(__file__), "..", "main.py")
        with open(_mp) as f:
            src = f.read()
        start = src.index("_ALLOWED_ORIGINS_ENV = os.getenv")
        def_idx = src.index("def _is_allowed_origin")
        end = src.index("\n\n", def_idx)
        block = src[start:end]
        env_ctx = unittest.mock.patch.dict(os.environ, extra_env or {})
        with env_ctx:
            ns: dict = {"os": os}
            class _StubLogger:
                def info(self, *a, **k): pass
                def warning(self, *a, **k): pass
            ns["_logger"] = _StubLogger()
            exec(compile(block, "main_cors_block", "exec"), ns)
        return ns["_is_allowed_origin"]

    def setUp(self):
        try:
            self._fn = self._load_fn()
        except Exception as e:
            self.skipTest(str(e))

    def test_blocks_unknown_origin(self):
        self.assertFalse(self._fn("https://evil.com"), "GAP-8: origin sconosciuto non bloccato")

    def test_blocks_none_origin(self):
        self.assertFalse(self._fn(None))

    def test_blocks_empty_origin(self):
        self.assertFalse(self._fn(""))

    def test_allows_localhost_dev(self):
        self.assertTrue(self._fn("http://localhost:5173"))

    def test_allows_pages_dev_pattern(self):
        self.assertTrue(self._fn("https://agente-ai.pages.dev"))

    def test_allows_hf_space_pattern(self):
        self.assertTrue(self._fn("https://baida98-ai.hf.space"))

    def test_blocks_suffix_spoofing_lookalike(self):
        """'*.hf.space.evil.com' non deve ingannare il check endswith."""
        self.assertFalse(self._fn("https://notreal.hf.space.evil.com"),
            "GAP-8: suffix spoofing bypassa endswith check")

    def test_blocks_prefix_lookalike(self):
        self.assertFalse(self._fn("https://my-hf.space.attacker.io"))

    def test_env_configured_origin_allowed(self):
        """ALLOWED_ORIGINS env aggiunge origin custom alla whitelist — ma non apre a tutto."""
        try:
            fn2 = self._load_fn({"ALLOWED_ORIGINS": "https://custom-domain.com"})
        except Exception as e:
            self.skipTest(str(e))
        self.assertTrue(fn2("https://custom-domain.com"))
        self.assertFalse(fn2("https://random-other.com"))


class TestEmailFromDomainAllowlist(unittest.TestCase):
    """SEC-9: _from_domain_allowed() blocca mittenti non autorizzati (GAP-9-fix)."""

    def setUp(self):
        try:
            from api.email import _from_domain_allowed
            self._fn = _from_domain_allowed
        except ImportError as e:
            self.skipTest(str(e))

    def test_allows_configured_domain(self):
        with unittest.mock.patch.dict(os.environ, {
            "RESEND_ALLOWED_DOMAINS": "mydomain.com",
            "RESEND_FROM_EMAIL": "",
        }):
            self.assertTrue(self._fn("noreply@mydomain.com"))

    def test_blocks_unconfigured_domain(self):
        with unittest.mock.patch.dict(os.environ, {
            "RESEND_ALLOWED_DOMAINS": "mydomain.com",
            "RESEND_FROM_EMAIL": "",
        }):
            self.assertFalse(self._fn("attacker@evil.com"),
                "GAP-9: open relay — dominio non autorizzato non bloccato")

    def test_uses_resend_from_email_fallback(self):
        with unittest.mock.patch.dict(os.environ, {
            "RESEND_ALLOWED_DOMAINS": "",
            "RESEND_FROM_EMAIL": "agent@myapp.io",
        }):
            self.assertTrue(self._fn("noreply@myapp.io"))
            self.assertFalse(self._fn("hack@attacker.net"),
                "GAP-9: dominio non in fallback RESEND_FROM_EMAIL non bloccato")

    def test_blocks_invalid_email_format(self):
        with unittest.mock.patch.dict(os.environ, {
            "RESEND_ALLOWED_DOMAINS": "x.com",
            "RESEND_FROM_EMAIL": "",
        }):
            self.assertFalse(self._fn("notanemail"))

    def test_blocks_empty_from(self):
        with unittest.mock.patch.dict(os.environ, {
            "RESEND_ALLOWED_DOMAINS": "x.com",
            "RESEND_FROM_EMAIL": "",
        }):
            self.assertFalse(self._fn(""))

    def test_multi_domain_allowlist(self):
        with unittest.mock.patch.dict(os.environ, {
            "RESEND_ALLOWED_DOMAINS": "domain-a.com,domain-b.org",
            "RESEND_FROM_EMAIL": "",
        }):
            self.assertTrue(self._fn("no-reply@domain-a.com"))
            self.assertTrue(self._fn("bot@domain-b.org"))
            self.assertFalse(self._fn("spy@domain-c.net"))


class TestTokenAutoRefreshBehavior(unittest.TestCase):
    """SEC-10: get_managed_token esegue davvero il refresh end-to-end (GAP-7-fix).

    Mocka Supabase (_sb_get_token/_sb_upsert_token) e httpx.AsyncClient per
    verificare comportamento reale, non solo presenza lessicale di token nel sorgente.
    """

    def setUp(self):
        try:
            import api.auth_managed as amod
            self.amod = amod
        except ImportError as e:
            self.skipTest(str(e))

    def _row(self, expires_at, access="ACCESS_ENC", refresh="REFRESH_ENC"):
        return {
            'access_token': access, 'refresh_token': refresh,
            'expires_at': expires_at, 'scope': 'read:user', 'raw_meta': '{}',
        }

    def test_valid_token_skips_refresh(self):
        """Token valido (>60s residui) → NON deve chiamare l'HTTP refresh."""
        far_future = int((time.time() + 3600) * 1000)
        row = self._row(far_future, access='PLAINTEXT_ACCESS')

        async def fake_get(user_id, provider): return row

        async def forbidden_post(*a, **k):
            raise AssertionError("GAP-7: refresh chiamato anche con token ancora valido")

        with unittest.mock.patch.object(self.amod, '_sb_get_token', fake_get), \
             unittest.mock.patch.object(self.amod, '_decrypt', lambda t: t), \
             unittest.mock.patch('httpx.AsyncClient.post', forbidden_post):
            result = _run(self.amod.get_managed_token('default', 'github'))
        self.assertEqual(result, 'PLAINTEXT_ACCESS')

    def test_expired_token_refresh_success_persists_new_token(self):
        """Token scaduto + refresh_token valido → POST al provider, salva e ritorna il nuovo access_token."""
        past = int((time.time() - 100) * 1000)
        row = self._row(past)
        upserted: dict = {}

        async def fake_get(user_id, provider): return row

        async def fake_upsert(user_id, provider, access, refresh, expires_at, scope, meta):
            upserted.update(access=access, refresh=refresh, expires_at=expires_at)

        class FakeResp:
            status_code = 200
            def json(self):
                return {'access_token': 'NEW_ACCESS', 'refresh_token': 'NEW_REFRESH', 'expires_in': 3600}

        async def fake_post(self, *a, **k):
            return FakeResp()

        with unittest.mock.patch.object(self.amod, '_sb_get_token', fake_get), \
             unittest.mock.patch.object(self.amod, '_sb_upsert_token', fake_upsert), \
             unittest.mock.patch.object(self.amod, '_decrypt', lambda t: t), \
             unittest.mock.patch('httpx.AsyncClient.post', fake_post), \
             unittest.mock.patch.dict(os.environ, {
                 'GITHUB_OAUTH_CLIENT_ID': 'cid', 'GITHUB_OAUTH_CLIENT_SECRET': 'csec'}):
            result = _run(self.amod.get_managed_token('default', 'github'))

        self.assertEqual(result, 'NEW_ACCESS', "GAP-7: refresh riuscito ma non ritorna il nuovo token")
        self.assertEqual(upserted.get('access'), 'NEW_ACCESS', "GAP-7: nuovo token non persistito su Supabase")

    def test_expired_token_refresh_fails_falls_back_to_old_token(self):
        """Refresh HTTP fallisce (status != 200) → ritorna comunque il vecchio access_token, nessun crash."""
        past = int((time.time() - 100) * 1000)
        row = self._row(past, access='OLD_ACCESS')

        async def fake_get(user_id, provider): return row

        class FakeResp:
            status_code = 400
            text = 'invalid_grant'
            def json(self): return {}

        async def fake_post(self, *a, **k):
            return FakeResp()

        with unittest.mock.patch.object(self.amod, '_sb_get_token', fake_get), \
             unittest.mock.patch.object(self.amod, '_decrypt', lambda t: t), \
             unittest.mock.patch('httpx.AsyncClient.post', fake_post), \
             unittest.mock.patch.dict(os.environ, {
                 'GITHUB_OAUTH_CLIENT_ID': 'cid', 'GITHUB_OAUTH_CLIENT_SECRET': 'csec'}):
            result = _run(self.amod.get_managed_token('default', 'github'))
        self.assertEqual(result, 'OLD_ACCESS')

    def test_no_refresh_token_returns_existing_without_crash(self):
        """Nessun refresh_token salvato → ritorna il token esistente anche se scaduto."""
        past = int((time.time() - 100) * 1000)
        row = self._row(past, refresh='')

        async def fake_get(user_id, provider): return row

        with unittest.mock.patch.object(self.amod, '_sb_get_token', fake_get), \
             unittest.mock.patch.object(self.amod, '_decrypt', lambda t: t):
            result = _run(self.amod.get_managed_token('default', 'github'))
        self.assertEqual(result, 'ACCESS_ENC')

    def test_unknown_or_disconnected_provider_returns_none(self):
        async def fake_get(user_id, provider): return None
        with unittest.mock.patch.object(self.amod, '_sb_get_token', fake_get):
            result = _run(self.amod.get_managed_token('default', 'github'))
        self.assertIsNone(result)


class TestOAuthStateBehavior(unittest.TestCase):
    """SEC-11: OAuth state — persistenza Supabase, TTL, one-shot, fail-closed (GAP-6-fix).

    Usa un fake client Supabase (query builder minimale) per eseguire il vero
    codice di _make_state/_consume_state end-to-end, senza rete reale.
    """

    class _FakeTable:
        def __init__(self, store):
            self.store = store
            self._eq = (None, None)
            self._delete = False

        def insert(self, payload):
            self.store.append(dict(payload))
            return self

        def select(self, *a, **k):
            return self

        def eq(self, field, value):
            self._eq = (field, value)
            return self

        def delete(self):
            self._delete = True
            return self

        def limit(self, n):
            return self

        def execute(self):
            field, value = self._eq
            if self._delete:
                self.store[:] = [r for r in self.store if r.get(field) != value]
                return unittest.mock.MagicMock(data=[])
            matches = [r for r in self.store if r.get(field) == value] if field else list(self.store)
            return unittest.mock.MagicMock(data=matches)

    class _FakeSupabase:
        def __init__(self):
            self._store: list = []

        def table(self, name):
            return TestOAuthStateBehavior._FakeTable(self._store)

    def setUp(self):
        try:
            import api.auth_managed as amod
            self.amod = amod
        except ImportError as e:
            self.skipTest(str(e))

    def test_make_state_persists_to_supabase(self):
        fake_sb = self._FakeSupabase()
        import api.state as state_mod
        with unittest.mock.patch.object(state_mod, '_sb', fake_sb):
            state = _run(self.amod._make_state('github', 'default'))
        self.assertEqual(len(fake_sb._store), 1, "GAP-6: _make_state non ha scritto su Supabase")
        self.assertEqual(fake_sb._store[0]['state'], state)
        self.assertEqual(fake_sb._store[0]['provider'], 'github')

    def test_consume_state_one_shot_no_replay(self):
        """Dopo il primo consumo lo stesso state deve fallire — impedisce replay CSRF."""
        fake_sb = self._FakeSupabase()
        import api.state as state_mod
        fake_sb._store.append({
            'state': 'abc123', 'provider': 'github', 'user_id': 'default',
            'created_at': datetime.now(timezone.utc).isoformat(),
        })
        with unittest.mock.patch.object(state_mod, '_sb', fake_sb):
            first = _run(self.amod._consume_state('abc123'))
            replay = _run(self.amod._consume_state('abc123'))
        self.assertIsNotNone(first)
        self.assertEqual(first['provider'], 'github')
        self.assertIsNone(replay, "GAP-6: state riutilizzabile — replay CSRF possibile")

    def test_consume_state_expired_ttl_rejected(self):
        """Uno state più vecchio di _STATE_TTL secondi deve essere rifiutato."""
        fake_sb = self._FakeSupabase()
        import api.state as state_mod
        old_time = datetime.now(timezone.utc) - timedelta(seconds=self.amod._STATE_TTL + 60)
        fake_sb._store.append({
            'state': 'old_state', 'provider': 'github', 'user_id': 'default',
            'created_at': old_time.isoformat(),
        })
        with unittest.mock.patch.object(state_mod, '_sb', fake_sb):
            result = _run(self.amod._consume_state('old_state'))
        self.assertIsNone(result, "GAP-6: state expired accettato — TTL non applicato")

    def test_consume_state_unknown_rejected(self):
        fake_sb = self._FakeSupabase()
        import api.state as state_mod
        with unittest.mock.patch.object(state_mod, '_sb', fake_sb):
            result = _run(self.amod._consume_state('never-existed'))
        self.assertIsNone(result)

    def test_consume_fails_closed_when_supabase_down_in_production(self):
        """Supabase irraggiungibile in produzione → rifiuta lo state (fail-closed), niente fallback insicuro."""
        class _BrokenTable:
            def select(self, *a, **k): return self
            def eq(self, *a, **k): return self
            def limit(self, *a, **k): return self
            def execute(self): raise ConnectionError("supabase down")

        class _BrokenSupabase:
            def table(self, name): return _BrokenTable()

        import api.state as state_mod
        with unittest.mock.patch.object(state_mod, '_sb', _BrokenSupabase()), \
             unittest.mock.patch.dict(os.environ, {'RAILWAY_ENVIRONMENT': 'production'}):
            result = _run(self.amod._consume_state('any-state'))
        self.assertIsNone(result,
            "GAP-6: fail-open in produzione con Supabase down — CSRF bypass possibile")


class TestWebhookSetWebhookAdminGate(unittest.TestCase):
    """SEC-12: /api/telegram/set-webhook fail-closed quando ADMIN_TOKEN non è configurato
    (GAP-WEBHOOK-ADMIN-FIX, audit 2026-07-09).

    Usa un vero TestClient FastAPI (non chiamata diretta alla dependency) perché
    require_role() ha una sotto-dependency annidata (_resolve_role) che va
    risolta dal framework — vedi nota in GAPS_TODO.md.
    """

    def setUp(self):
        try:
            from fastapi import FastAPI
            from fastapi.testclient import TestClient
            import api.webhook as webhook_mod
            app = FastAPI()
            app.include_router(webhook_mod.router)
            self.client = TestClient(app)
        except ImportError as e:
            self.skipTest(str(e))

    def test_rejects_without_admin_token_configured(self):
        """ADMIN_TOKEN non settato → 503 (fail-closed), MAI 200."""
        with unittest.mock.patch.dict(os.environ, {"ADMIN_TOKEN": ""}, clear=False):
            resp = self.client.post("/api/telegram/set-webhook")
        self.assertEqual(resp.status_code, 503,
            "GAP-WEBHOOK-ADMIN-FIX: endpoint apribile senza ADMIN_TOKEN configurato")

    def test_rejects_wrong_admin_token(self):
        """ADMIN_TOKEN configurato ma header errato/assente → 403."""
        with unittest.mock.patch.dict(os.environ, {"ADMIN_TOKEN": "correct-secret"}, clear=False):
            resp = self.client.post("/api/telegram/set-webhook",
                                     headers={"X-Admin-Token": "wrong-guess"})
        self.assertEqual(resp.status_code, 403)

    def test_no_admin_header_rejected_even_if_token_configured(self):
        with unittest.mock.patch.dict(os.environ, {"ADMIN_TOKEN": "correct-secret"}, clear=False):
            resp = self.client.post("/api/telegram/set-webhook")
        self.assertIn(resp.status_code, (401, 403))


if __name__ == "__main__":
    unittest.main()