|
| 1 | +"""Additional tests to reach 80%+ coverage.""" |
| 2 | +import os |
| 3 | +import sys |
| 4 | +import tempfile |
| 5 | +import json |
| 6 | +from io import BytesIO |
| 7 | +from unittest.mock import patch, MagicMock |
| 8 | + |
| 9 | +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
| 10 | + |
| 11 | +import pytest |
| 12 | +from localportmanager import LocalPortManager, PortRegistry, ReverseProxyHandler, main |
| 13 | + |
| 14 | + |
| 15 | +class TestAdditionalCoverage: |
| 16 | + """Additional tests for better coverage.""" |
| 17 | + |
| 18 | + def test_register_service_with_yes_flag(self): |
| 19 | + """Test register command with --yes flag.""" |
| 20 | + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: |
| 21 | + state_file = f.name |
| 22 | + |
| 23 | + try: |
| 24 | + with patch('os.system') as mock_system, \ |
| 25 | + patch('sys.argv', [ |
| 26 | + 'localportmanager', |
| 27 | + '--state-file', state_file, |
| 28 | + 'register', |
| 29 | + 'test-service', |
| 30 | + 'echo {port}', |
| 31 | + '--yes' |
| 32 | + ]): |
| 33 | + main() |
| 34 | + |
| 35 | + # Auto-start with --yes should call os.system |
| 36 | + mock_system.assert_called_once() |
| 37 | + finally: |
| 38 | + if os.path.exists(state_file): |
| 39 | + os.unlink(state_file) |
| 40 | + |
| 41 | + def test_port_registry_list_services_returns_copy(self): |
| 42 | + """Test that list_services returns a copy.""" |
| 43 | + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: |
| 44 | + state_file = f.name |
| 45 | + |
| 46 | + try: |
| 47 | + registry = PortRegistry(state_file=state_file) |
| 48 | + registry.register("test", 4001) |
| 49 | + services = registry.list_services() |
| 50 | + services["new"] = 4002 |
| 51 | + |
| 52 | + # Original should be unchanged |
| 53 | + assert "new" not in registry.list_services() |
| 54 | + finally: |
| 55 | + if os.path.exists(state_file): |
| 56 | + os.unlink(state_file) |
| 57 | + |
| 58 | + def test_localportmanager_init_with_none_state_file(self): |
| 59 | + """Test LocalPortManager init with None state_file.""" |
| 60 | + lpm = LocalPortManager(proxy_port=1355, state_file=None) |
| 61 | + assert lpm.registry is not None |
| 62 | + assert lpm.registry.state_file == "/tmp/localportmanager_registry.json" |
| 63 | + |
| 64 | + def test_proxy_handler_with_path_routing(self): |
| 65 | + """Test proxy handler path-based routing.""" |
| 66 | + handler = MagicMock() |
| 67 | + handler.headers = {'Host': '127.0.0.1:1355'} |
| 68 | + handler.path = '/api/test' |
| 69 | + handler.rfile = BytesIO(b'') |
| 70 | + |
| 71 | + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: |
| 72 | + state_file = f.name |
| 73 | + f.write('{"api": 4001}') |
| 74 | + |
| 75 | + try: |
| 76 | + # Mock send_error to avoid 404 |
| 77 | + from localportmanager import ReverseProxyHandler |
| 78 | + ReverseProxyHandler._proxy_request(handler, 'GET') |
| 79 | + finally: |
| 80 | + if os.path.exists(state_file): |
| 81 | + os.unlink(state_file) |
| 82 | + |
| 83 | + def test_main_register_invalid_name(self): |
| 84 | + """Test register with invalid name shows error.""" |
| 85 | + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: |
| 86 | + state_file = f.name |
| 87 | + |
| 88 | + try: |
| 89 | + with patch('sys.argv', [ |
| 90 | + 'localportmanager', |
| 91 | + '--state-file', state_file, |
| 92 | + 'register', |
| 93 | + 'invalid name!', |
| 94 | + 'cmd' |
| 95 | + ]): |
| 96 | + with pytest.raises(SystemExit) as exc_info: |
| 97 | + main() |
| 98 | + |
| 99 | + assert exc_info.value.code == 1 |
| 100 | + finally: |
| 101 | + if os.path.exists(state_file): |
| 102 | + os.unlink(state_file) |
| 103 | + |
| 104 | + def test_register_service_output_format(self, capsys): |
| 105 | + """Test register service output format.""" |
| 106 | + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: |
| 107 | + state_file = f.name |
| 108 | + |
| 109 | + try: |
| 110 | + lpm = LocalPortManager(proxy_port=8888, state_file=state_file) |
| 111 | + |
| 112 | + with patch('builtins.input', return_value='n'): |
| 113 | + lpm.register_service("myapp", "python app.py {port}") |
| 114 | + |
| 115 | + captured = capsys.readouterr() |
| 116 | + |
| 117 | + assert "myapp" in captured.out |
| 118 | + assert "127.0.0.1" in captured.out |
| 119 | + assert "localhost" in captured.out |
| 120 | + assert "python app.py" in captured.out |
| 121 | + finally: |
| 122 | + if os.path.exists(state_file): |
| 123 | + os.unlink(state_file) |
| 124 | + |
| 125 | + def test_status_command_with_services(self, capsys): |
| 126 | + """Test status command output with services.""" |
| 127 | + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: |
| 128 | + state_file = f.name |
| 129 | + f.write('{"webapp": 4001, "api": 4002}') |
| 130 | + |
| 131 | + try: |
| 132 | + with patch('sys.argv', [ |
| 133 | + 'localportmanager', |
| 134 | + '--port', '8888', |
| 135 | + '--state-file', state_file, |
| 136 | + 'status' |
| 137 | + ]): |
| 138 | + main() |
| 139 | + |
| 140 | + captured = capsys.readouterr() |
| 141 | + |
| 142 | + assert "8888" in captured.out |
| 143 | + assert "webapp" in captured.out |
| 144 | + assert "api" in captured.out |
| 145 | + finally: |
| 146 | + if os.path.exists(state_file): |
| 147 | + os.unlink(state_file) |
0 commit comments