"""Targeted local tool tests. No server connection, credentials, database or device writes.""" import json from pathlib import Path import socket import select import tempfile import time import unittest from types import SimpleNamespace from unittest.mock import patch import qa import qa_isolation as isolation class QaIsolationTest(unittest.TestCase): def item(self): return {'state':'READY','database':isolation.DATABASE,'port':15432,'heartbeat':time.time(), 'isolation_id':'synthetic','owner':{'pid':101},'ssh':{'pid':102}} def test_legacy_url_is_refused_before_database_access(self): for host in ['127.0.0.1','localhost','192.168.139.218']: with self.assertRaises(ValueError): qa.environment({'FUSION_QA_DB_URL':'jdbc:kingbase8://'+host+':14321/'+qa.DATABASE, 'FUSION_QA_APP_PASSWORD':'synthetic'}) def test_no_lease_prevents_even_database_preflight(self): with patch.object(isolation,'require',side_effect=ValueError('synthetic refusal')), \ patch.object(qa,'inspect') as inspect, patch.object(qa,'launch') as launch: with self.assertRaises(ValueError):qa.start({}, {}, 18096, 16389) inspect.assert_not_called();launch.assert_not_called() def test_ready_owned_transport_allowed(self): item=self.item() with patch.object(isolation,'same_process',return_value=True), \ patch.object(isolation,'listener_owned',return_value=True), \ patch.object(isolation,'conflicting_clients',return_value=[]): self.assertEqual(item,isolation.validate(item,isolation.url_for(item))) def test_expired_or_future_heartbeat_refused(self): for delta in [-11,60]: item=self.item();item['heartbeat']=time.time()+delta with self.assertRaises(ValueError):isolation.validate(item) def test_legacy_and_invalid_ports_refused(self): for port in [14321,0,-1,65536,'15432']: item=self.item();item['port']=port with self.assertRaises(ValueError):isolation.validate(item) def test_wrong_state_or_database_refused(self): for key,value in [('state','STOPPED'),('database','training')]: item=self.item();item[key]=value with self.assertRaises(ValueError):isolation.validate(item) def test_reused_pid_refused(self): with patch.object(isolation,'same_process',return_value=False): with self.assertRaises(ValueError):isolation.validate(self.item()) def test_unknown_listener_refused(self): with patch.object(isolation,'same_process',return_value=True),patch.object(isolation,'listener_owned',return_value=False): with self.assertRaises(ValueError):isolation.validate(self.item()) def test_wrong_application_endpoint_refused(self): for url in ['jdbc:kingbase8://127.0.0.1:14321/'+qa.DATABASE, 'jdbc:kingbase8://192.168.139.218:4321/'+qa.DATABASE, 'jdbc:kingbase8://127.0.0.1:15432/training']: with patch.object(isolation,'same_process',return_value=True),patch.object(isolation,'listener_owned',return_value=True): with self.assertRaises(ValueError):isolation.validate(self.item(),url) def test_another_java_client_is_not_allowed_to_share_endpoint(self): with patch.object(isolation,'same_process',return_value=True), \ patch.object(isolation,'listener_owned',return_value=True), \ patch.object(isolation,'conflicting_clients',return_value=[999]): with self.assertRaises(ValueError):isolation.validate(self.item()) def test_deny_socket_has_no_upstream_and_rejects_data(self): # Use an OS-selected temporary port, never touch the actual legacy port in unit tests. with patch.object(isolation,'LEGACY_PORT',0):guard=isolation.deny_socket() try: with socket.create_connection(guard.getsockname(),timeout=1) as client: self.assertTrue(select.select([guard],[],[],2)[0]) peer,_=guard.accept();peer.close() self.assertEqual(b'',client.recv(1)) finally:guard.close() def test_existing_listener_not_replaced(self): with socket.socket() as listener: listener.bind(('127.0.0.1',0));listener.listen(1) with patch.object(isolation,'LEGACY_PORT',listener.getsockname()[1]): with self.assertRaises(OSError):isolation.deny_socket() self.assertNotEqual(-1,listener.fileno()) def test_tunnel_never_reuses_a_controlmaster_or_exposes_password(self): command=isolation.ssh_command(15432) self.assertIn('ControlMaster=no',command);self.assertIn('ControlPath=none',command) self.assertEqual('/dev/null',command[command.index('-F')+1]) self.assertTrue(any(p.startswith('ProxyCommand=') and 'ClearAllForwardings=yes' in p for p in command)) self.assertIn('ExitOnForwardFailure=yes',command) self.assertIn('-L127.0.0.1:15432:127.0.0.1:4321',command) self.assertNotIn('-f',command);self.assertNotIn('-g',command) self.assertFalse(any('password=' in p.lower() for p in command)) def test_lease_file_is_private_and_symlink_refused(self): with tempfile.TemporaryDirectory() as directory,patch.object(isolation,'RUNTIME',Path(directory)): isolation.write_lease({'synthetic':True});path=Path(directory)/isolation.LEDGER self.assertEqual(0o600,path.stat().st_mode & 0o777) self.assertEqual({'synthetic':True},isolation.read_lease()) path.unlink();path.symlink_to(Path(directory)/'other') with self.assertRaises((ValueError,OSError)):isolation.read_lease() def test_stop_cannot_signal_another_lease(self): with tempfile.TemporaryDirectory() as directory,patch.object(qa,'RUNTIME',Path(directory)),patch.object(qa.os,'kill') as kill: (Path(directory)/'backend.json').write_text(json.dumps({'pid':123,'isolation_id':'other'})) with self.assertRaises(ValueError):qa.stop(isolation_id='synthetic') kill.assert_not_called() def test_foreign_client_scan_does_not_report_environment_secrets(self): def fake(args): if args[:2]==['ps','-axo']:return SimpleNamespace(stdout='123 /usr/bin/java\n',returncode=0) return SimpleNamespace(stdout='java DB_PORT=15432 SOME_PASSWORD=synthetic_private_value',returncode=0) with patch.object(isolation,'run',side_effect=fake): self.assertEqual([123],isolation.conflicting_clients(15432,'synthetic')) def test_frontend_cannot_proxy_an_old_backend_with_matching_lease_label(self): with tempfile.TemporaryDirectory() as directory,patch.object(qa,'RUNTIME',Path(directory)), \ patch.object(isolation,'require',return_value={'isolation_id':'synthetic'}), \ patch.object(qa,'launch') as launch,patch('sys.argv',['qa.py','frontend','--backend-port','8096']): (Path(directory)/'backend.json').write_text(json.dumps({'isolation_id':'synthetic','command':['java','--server.port=18096']})) with self.assertRaises(ValueError):qa.main() launch.assert_not_called() if __name__=='__main__':unittest.main()