business-assistant/scripts/tests/test_dms_connection.py

import contextlib
import io
import json
import os
from pathlib import Path
import socket
import tempfile
import unittest
from unittest.mock import patch
from urllib.error import URLError, HTTPError

from scripts import test_dms_connection as probe
from src.dms_schema import DEFAULT_SCHEMA


class DiagnosticTests(unittest.TestCase):
    def test_compact_output_keeps_failures_and_hides_matched_fields(self):
        self.mismatch = True
        with patch('urllib.request.OpenerDirector.open', side_effect=self.fake), contextlib.redirect_stderr(io.StringIO()):
            report, _ = probe.run(self.root, ('records', 'honor'), env_file_only=True)
        short = probe.compact_report(report, 2)
        self.assertEqual(short['passed_columns'], 1)
        self.assertEqual(short['failed_columns'], 1)
        self.assertNotIn('honor', short['failed_checks'])
        self.assertEqual({f['status'] for f in short['failed_checks']['records']['fields']}, {'missing', 'type_mismatch'})
        self.assertEqual(len(short['related_requests']), 1)
        self.assertIn('getModelById', short['related_requests'][0]['endpoint'])
        early = probe.compact_report({'ok': False, 'stage': 'authentication', 'code': 'dms_login_failed',
            'checks': {}, 'requests': [{'http_status': 404, 'error': 'dms_http_error'}]}, 5)
        self.assertEqual(early['not_checked_columns'], 5)
        self.assertEqual(early['related_requests'][0]['http_status'], 404)

    def test_cli_default_quiet_and_verbose_opt_in(self):
        def fake_run(*args, **kwargs):
            probe.emit('request_start', {'endpoint': 'synthetic'})
            return {'ok': True, 'checks': {'records': {'ok': True, 'fields': [{'status': 'matched'}]}}, 'requests': []}, set()
        for verbose in (False, True):
            out, err = io.StringIO(), io.StringIO()
            argv = ['test_dms_connection.py', '--column', 'records'] + (['--verbose'] if verbose else [])
            with patch.object(probe, 'run', side_effect=fake_run), patch('sys.argv', argv), \
                    contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
                self.assertEqual(probe.main(), 0)
            result = json.loads(out.getvalue())
            if verbose:
                self.assertIn('request_start', err.getvalue())
                self.assertIn('checks', result)
            else:
                self.assertEqual(err.getvalue(), '')
                self.assertEqual(result, {'ok': True, 'checked_columns': 1, 'passed_columns': 1,
                                          'failed_columns': 0, 'not_checked_columns': 0})

    def test_command_line_overrides_with_old_config_without_login_url(self):
        from types import SimpleNamespace
        original_loader = probe.load_dms_config
        def legacy_loader(*args, **kwargs):
            config = original_loader(*args, **kwargs)
            values = {k: v for k, v in vars(config).items() if k != 'login_url'}
            return SimpleNamespace(**values, check_scope=config.check_scope, alias_for=config.alias_for)
        def fake(request, timeout):
            if request.full_url == 'http://auth.invalid:8888/oauth/api/user/login':
                return Response({'code': 200, 'message': 'new-private-token'})
            self.assertTrue(request.full_url.startswith('http://data.invalid:10081/dms/'))
            return self.fake(request, timeout)
        (self.root / '.env').write_text('DMS_BASE_URL=http://localhost\nAUTH_PORT=9999\nDMS_PORT=9998\n'
                                      'DMS_USERNAME=private-user\nDMS_PASSWORD=private-password\n')
        original = (self.root / '.env').read_bytes()
        with patch.object(probe, 'load_dms_config', side_effect=legacy_loader), \
                patch('src.step5_session_analysis.dms.connection_urls',
                      return_value=('http://localhost:9998/proxy_dms', 'http://localhost:9999/proxy_oauth/user/login')), \
                patch('urllib.request.OpenerDirector.open', side_effect=fake), contextlib.redirect_stderr(io.StringIO()):
            report, _ = probe.run(self.root, ('records',), env_file_only=True,
                api_url='http://data.invalid:10081/dms', login_url='http://auth.invalid:8888/oauth/api/user/login')
        self.assertTrue(report['ok'], report)
        self.assertEqual(report['login_url_source'], 'command_line')
        self.assertEqual((self.root / '.env').read_bytes(), original)

    def test_invalid_override_never_sends_request(self):
        for url in ('http://user:secret@invalid/login', 'http://invalid/login?token=secret',
                    'http://invalid:99999/login', 'http://invalid/a b', 'relative', ''):
            with patch('urllib.request.OpenerDirector.open') as send, contextlib.redirect_stderr(io.StringIO()):
                report, _ = probe.run(self.root, ('records',), login_url=url)
            self.assertEqual(report['code'], 'invalid_endpoint_override')
            send.assert_not_called()

    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        (self.root / 'config').mkdir()
        self.schema = json.loads(json.dumps(DEFAULT_SCHEMA))
        (self.root / 'config/dms_schema.json').write_text(json.dumps(self.schema), encoding='utf-8')
        (self.root / '.env').write_text('DMS_BASE_URL=http://localhost:1234\nDMS_TOKEN=synthetic-private-token\n', encoding='utf-8')
        self.forms = []
        self.mismatch = False

    def fake(self, request, timeout):
        from urllib.parse import parse_qs
        form = parse_qs(request.data.decode())
        self.forms.append((request.full_url, form))
        if request.full_url.endswith('getColumnById'):
            identity = int(form['id'][0])
            spec = next((v for v in self.schema['columns'].values() if v['id'] == identity), None)
            content = {'title': spec['name'], 'type': 1, 'parentId': 999, 'modelId': identity} if spec else {'title': self.schema['parent']}
        else:
            spec = next(v for v in self.schema['columns'].values() if v['id'] == int(form['modelId'][0]))
            raw = {actual: {'type': spec.get('types', {}).get(logical, 'text')} for logical, actual in spec['fields'].items()}
            if self.mismatch:
                raw.pop('c_question', None)
                if 'c_created_at' in raw:
                    raw['c_created_at']['type'] = 'text'
            content = {'fieldList': json.dumps(raw)}
        return Response({'code': 200, 'content': content})

    def test_all_columns_requests_and_preflight_logs(self):
        logs = io.StringIO()
        with patch('urllib.request.OpenerDirector.open', side_effect=self.fake), contextlib.redirect_stderr(logs):
            report, secrets = probe.run(self.root, probe.ALIASES, env_file_only=True)
        self.assertTrue(report['ok'], report)
        self.assertEqual(len(self.forms), 15)
        self.assertTrue(all('/content/' not in url for url, _ in self.forms))
        self.assertIn('connection_parameters', logs.getvalue())
        self.assertIn('column_start', logs.getvalue())
        self.assertIn('"id":', logs.getvalue())
        self.assertNotIn('synthetic-private-token', logs.getvalue() + probe.redact_report(report, secrets))

    def test_missing_and_wrong_field_types_are_specific(self):
        self.mismatch = True
        with patch('urllib.request.OpenerDirector.open', side_effect=self.fake), contextlib.redirect_stderr(io.StringIO()):
            report, _ = probe.run(self.root, ('records',), env_file_only=True)
        check = report['checks']['records']
        self.assertFalse(check['ok'])
        fields = {v['logical_field']: v for v in check['fields']}
        self.assertEqual(fields['c_question']['status'], 'missing')
        self.assertEqual(fields['c_created_at']['status'], 'type_mismatch')
        self.assertEqual(fields['c_created_at']['expected_type'], 'timestamp')

    def test_scope_rejection_makes_no_requests(self):
        with (self.root / '.env').open('a') as f:
            f.write('DMS_STRICT_SCOPE=true\nDMS_ALLOWED_SCOPES_JSON={}\n')
        with patch('urllib.request.OpenerDirector.open') as send, contextlib.redirect_stderr(io.StringIO()):
            report, _ = probe.run(self.root, ('records',), env_file_only=True)
        self.assertEqual(report['code'], 'dms_scope_denied')
        send.assert_not_called()

    def test_network_http_and_auth_failures_keep_diagnostics(self):
        for error, expected in [(URLError(socket.gaierror(-2, 'secret-error')), 'dns_failed'),
                                (URLError(ConnectionRefusedError(111, 'secret-error')), 'connection_refused'),
                                (URLError(TimeoutError('secret-error')), 'timeout')]:
            with patch('urllib.request.OpenerDirector.open', side_effect=error), contextlib.redirect_stderr(io.StringIO()):
                report, _ = probe.run(self.root, ('records',), env_file_only=True)
            self.assertEqual(report['requests'][0]['error'], expected)
            self.assertNotIn('secret-error', json.dumps(report))
        error = HTTPError('http://localhost', 404, 'secret-error', {}, io.BytesIO(b'secret-body'))
        with patch('urllib.request.OpenerDirector.open', side_effect=error), contextlib.redirect_stderr(io.StringIO()):
            report, _ = probe.run(self.root, ('records',), env_file_only=True)
        self.assertEqual(report['requests'][0]['http_status'], 404)
        self.assertNotIn('secret-body', json.dumps(report))
        with patch('urllib.request.OpenerDirector.open', return_value=Response({'code': 212})), contextlib.redirect_stderr(io.StringIO()):
            report, _ = probe.run(self.root, ('records',), env_file_only=True)
        self.assertEqual(report['checks']['records']['code'], 'dms_token_invalid')

    def test_login_values_and_returned_token_never_printed(self):
        (self.root / '.env').write_text('DMS_BASE_URL=http://localhost:1234/dms\nDMS_LOGIN_URL=http://localhost:8888/oauth/api/user/login\nDMS_USERNAME=private-user\nDMS_PASSWORD=private-password\n')
        def fake(request, timeout):
            if request.full_url.endswith('/user/login'):
                self.assertEqual(request.full_url, 'http://localhost:8888/oauth/api/user/login')
                return Response({'code': 200, 'message': 'fresh-private-token'})
            return self.fake(request, timeout)
        logs = io.StringIO()
        with patch('urllib.request.OpenerDirector.open', side_effect=fake), contextlib.redirect_stderr(logs):
            report, secrets = probe.run(self.root, ('records',), env_file_only=True)
        self.assertTrue(report['ok'])
        self.assertEqual(report['api_endpoint'], 'http://localhost:1234/dms')
        text = logs.getvalue() + probe.redact_report(report, secrets)
        for secret in ('private-user', 'private-password', 'fresh-private-token'):
            self.assertNotIn(secret, text)
        self.assertIn('"serviceId": "0"', text)


class Response:
    status = 200
    length = None
    headers = {'Content-Type': 'application/json'}
    def __init__(self, payload):
        self.payload = payload
    def __enter__(self):
        return self
    def __exit__(self, *_):
        pass
    def read(self, limit):
        return json.dumps(self.payload).encode()[:limit]