"""Protocol and process-boundary checks for the deployed read-only diagnostic reader."""
import datetime
import importlib.util
import json
from pathlib import Path
import sys
import unittest
from unittest.mock import patch

SPEC = importlib.util.spec_from_file_location('reader', Path(__file__).with_name('identity_diagnostics.py'))
READER = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(READER)
SINCE = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(minutes=5)).strftime('%Y-%m-%dT%H:%M:%SZ')
CID = b'a' * 64


class IdentityDiagnostics(unittest.TestCase):
    def test_exact_labels_only(self):
        raw = b'\n'.join([
            b'firebase_identity_refused stage=auth_signature reason=token_invalid',
            b'WARNING:cast_tv.firebase:firebase_identity_refused stage=account_lookup reason=account_revoked',
            b'WARNING:second_product.identity:firebase_identity_refused stage=app_check_signature reason=unknown_key',
            b'Bearer PRIVATE_SENTINEL',
            b'firebase_identity_refused stage=PRIVATE_SENTINEL reason=token_invalid',
            b'firebase_identity_refused stage=auth_signature reason=PRIVATE_SENTINEL',
            b'firebase_identity_refused stage=auth_signature reason=token_invalid user=PRIVATE_SENTINEL',
            b'prefix firebase_identity_refused stage=auth_signature reason=token_invalid',
        ])
        self.assertEqual(READER.records(raw), [
            {'stage': 'auth_signature', 'reason': 'token_invalid'},
            {'stage': 'account_lookup', 'reason': 'account_revoked'},
            {'stage': 'app_check_signature', 'reason': 'unknown_key'},
        ])

    def test_recent_rows_bounded(self):
        self.assertEqual(len(READER.records(
            b'firebase_identity_refused stage=app_check_signature reason=unknown_key\n' * 1001)), 100)

    def test_invalid_scopes_and_windows(self):
        for app, service, since in [('../other', 'api', SINCE), ('app', 'api;id', SINCE),
                                    ('app', 'api', '2020-01-01T00:00:00Z'),
                                    ('app', 'api', '2099-01-01T00:00:00Z'), ('app', 'api', '10m')]:
            with self.subTest(app=app, service=service, since=since), self.assertRaises(READER.ReaderFailure):
                READER.validate(app, service, since)

    def test_remote_exact_container_recheck_and_fixed_projection(self):
        labels = json.dumps({'com.docker.compose.project': 'fixture', 'com.docker.compose.service': 'api'}).encode()
        with patch.object(READER, 'bounded', side_effect=[CID + b'\n', labels,
                b'PRIVATE_SENTINEL\nfirebase_identity_refused stage=app_scope reason=app_scope_invalid\n']) as run:
            result = READER.remote('fixture', 'api', SINCE)
        self.assertEqual(result['records'], [{'stage': 'app_scope', 'reason': 'app_scope_invalid'}])
        commands = [call.args[0] for call in run.call_args_list]
        self.assertEqual(commands[0], ['docker', 'ps', '--no-trunc', '--filter',
            'label=com.docker.compose.project=fixture', '--filter', 'label=com.docker.compose.service=api',
            '--format', '{{.ID}}'])
        self.assertEqual(commands[1][-1], CID.decode())
        self.assertEqual(commands[2][4:], ['docker', 'logs', '--since', SINCE, '--tail', '1000', CID.decode()])
        self.assertNotIn('PRIVATE_SENTINEL', json.dumps(result))

    def test_absent_ambiguous_and_foreign_containers_refuse(self):
        for ids in [b'', CID + b'\n' + CID, b'PRIVATE_SENTINEL']:
            with patch.object(READER, 'bounded', return_value=ids) as run, self.assertRaises(READER.ReaderFailure):
                READER.remote('fixture', 'api', SINCE)
            self.assertEqual(run.call_count, 1)
        for project, service in [('other', 'api'), ('fixture', 'db')]:
            labels = json.dumps({'com.docker.compose.project': project, 'com.docker.compose.service': service})
            with patch.object(READER, 'bounded', side_effect=[CID, labels]) as run, \
                    self.assertRaises(READER.ReaderFailure):
                READER.remote('fixture', 'api', SINCE)
            self.assertEqual(run.call_count, 2)

    def test_local_rejects_raw_or_extended_remote_output(self):
        args = type('Args', (), dict(app='fixture', service='api', since=SINCE, host='fixture.invalid', user='root'))()
        for raw in [b'PRIVATE_SENTINEL', json.dumps({'schema': READER.SCHEMA, 'ok': True,
                'records': [{'stage': 'auth_signature', 'reason': 'PRIVATE_SENTINEL'}]}).encode(),
                json.dumps({'schema': READER.SCHEMA, 'ok': True, 'records': [], 'raw': 'PRIVATE_SENTINEL'}).encode()]:
            with patch.object(READER, 'bounded', return_value=(0, raw)), self.assertRaises(READER.ReaderFailure):
                READER.local(args)

    def test_owned_subprocess_success_and_stderr_suppression(self):
        self.assertEqual(READER.bounded([sys.executable, '-c',
            'import sys; print("ok"); print("PRIVATE_SENTINEL", file=sys.stderr)']), b'ok\n')

    def test_failed_subprocess_does_not_export_output(self):
        with self.assertRaisesRegex(READER.ReaderFailure, '^command_failed$'):
            READER.bounded([sys.executable, '-c', 'print("PRIVATE_SENTINEL"); raise SystemExit(7)'])

    def test_output_limit_and_owned_pipe_cleanup(self):
        with self.assertRaisesRegex(READER.ReaderFailure, '^output_limit$'):
            READER.bounded([sys.executable, '-c', 'print("x" * 1048577)'])
        with self.assertRaisesRegex(READER.ReaderFailure, '^command_timeout$'):
            READER.bounded([sys.executable, '-c',
                'import os,time; child=os.fork(); time.sleep(10) if child == 0 else None'], seconds=0.2)

    def test_stdin_upload_is_bounded_and_complete(self):
        self.assertEqual(READER.bounded([sys.executable, '-c',
            'import sys; print(len(sys.stdin.buffer.read()))'], payload=b'x' * 200000), b'200000\n')


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