133 lines
7.2 KiB
Python
133 lines
7.2 KiB
Python
"""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()
|