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]