131 lines
5.5 KiB
Python
131 lines
5.5 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import http.client
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import ssl
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest import mock
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
SCRIPT = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "capabilities/am516-delivery-prediction/scripts/delivery_prediction_query_template.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location("delivery_query", SCRIPT)
|
|
query = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(query)
|
|
|
|
API_URL = "https://api.zeroerr-agent.com/api/SaleAgent/DeliveryPrediction/WithTransit"
|
|
|
|
|
|
class DeliveryNetworkTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.env = mock.patch.dict(os.environ, {}, clear=True)
|
|
self.env.start()
|
|
self.addCleanup(self.env.stop)
|
|
self.proxies = mock.patch.object(
|
|
urllib.request, "getproxies", return_value={"https": "http://127.0.0.1:7897"}
|
|
)
|
|
self.getproxies = self.proxies.start()
|
|
self.addCleanup(self.proxies.stop)
|
|
connection_patch = mock.patch.object(http.client, "HTTPSConnection")
|
|
self.connection = connection_patch.start()
|
|
self.addCleanup(connection_patch.stop)
|
|
response = self.connection.return_value.getresponse.return_value
|
|
response.status = response.code = 200
|
|
response.reason = "OK"
|
|
response.headers = {}
|
|
response.read.return_value = b'{"success": true, "data": {}}'
|
|
response.__enter__.return_value = response
|
|
|
|
def test_default_direct_ignores_system_and_environment_proxy(self) -> None:
|
|
os.environ["https_proxy"] = "http://127.0.0.1:9999"
|
|
before = dict(os.environ)
|
|
global_opener = urllib.request._opener
|
|
|
|
query.build_api_opener(API_URL).open(API_URL, timeout=30)
|
|
|
|
self.assertEqual(self.connection.call_args.args[0], "api.zeroerr-agent.com")
|
|
self.connection.return_value.set_tunnel.assert_not_called()
|
|
self.getproxies.assert_not_called()
|
|
self.assertEqual(dict(os.environ), before)
|
|
self.assertIs(urllib.request._opener, global_opener)
|
|
context = self.connection.call_args.kwargs["context"]
|
|
self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED)
|
|
self.assertTrue(context.check_hostname)
|
|
|
|
def test_system_mode_explicitly_uses_configured_proxy(self) -> None:
|
|
os.environ[query.PROXY_MODE_ENV] = "system"
|
|
with mock.patch.object(urllib.request, "proxy_bypass", return_value=False):
|
|
query.build_api_opener(API_URL).open(API_URL, timeout=30)
|
|
|
|
self.assertEqual(self.connection.call_args.args[0], "127.0.0.1:7897")
|
|
self.assertEqual(
|
|
self.connection.return_value.set_tunnel.call_args.args[0], "api.zeroerr-agent.com"
|
|
)
|
|
self.assertEqual(self.connection.call_args.kwargs["context"].verify_mode, ssl.CERT_REQUIRED)
|
|
|
|
def test_invalid_proxy_mode_stops_before_network_and_does_not_echo_value(self) -> None:
|
|
os.environ[query.PROXY_MODE_ENV] = "SECRET_MARKER"
|
|
with self.assertRaises(ValueError) as caught:
|
|
query.build_api_opener(API_URL)
|
|
self.assertNotIn("SECRET_MARKER", str(caught.exception))
|
|
self.connection.assert_not_called()
|
|
|
|
def test_direct_route_is_limited_to_approved_https_host(self) -> None:
|
|
for url in ("https://example.invalid/", "http://api.zeroerr-agent.com/",
|
|
"https://api.zeroerr-agent.com.example.invalid/"):
|
|
with self.subTest(url=url), self.assertRaises(ValueError):
|
|
query.build_api_opener(url)
|
|
self.connection.assert_not_called()
|
|
|
|
def run_main(self, *, skip_tls_verify: bool = False) -> tuple[int, dict]:
|
|
output = io.StringIO()
|
|
with (
|
|
mock.patch.object(query, "parse_args", return_value=SimpleNamespace(
|
|
product="eRob110H160I-FHM-18ET[V6]", quantity=10,
|
|
skip_tls_verify=skip_tls_verify,
|
|
)),
|
|
mock.patch.object(query, "get_api_key", return_value="test-key-not-a-real-secret"),
|
|
mock.patch.object(query, "DEFAULT_BASE_URL", "https://api.zeroerr-agent.com"),
|
|
contextlib.redirect_stdout(output),
|
|
):
|
|
code = query.main()
|
|
return code, json.loads(output.getvalue())
|
|
|
|
def test_controlled_script_sends_original_quantity_once_over_verified_direct_route(self) -> None:
|
|
code, payload = self.run_main()
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(payload["success"])
|
|
self.assertEqual(self.connection.call_args.args[0], "api.zeroerr-agent.com")
|
|
request = self.connection.return_value.request
|
|
request.assert_called_once()
|
|
method, path = request.call_args.args[:2]
|
|
self.assertEqual(method, "GET")
|
|
self.assertEqual(urllib.parse.parse_qs(urllib.parse.urlsplit(path).query), {
|
|
"product_specification": ["eRob110H160I-FHM-18ET[V6]"], "quantity": ["10"],
|
|
})
|
|
|
|
def test_certificate_failure_does_not_retry_or_fall_back_to_proxy(self) -> None:
|
|
self.connection.return_value.request.side_effect = ssl.SSLCertVerificationError(
|
|
1, "CERTIFICATE_VERIFY_FAILED"
|
|
)
|
|
code, payload = self.run_main()
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(payload["error"], "TLS certificate verification failed")
|
|
self.connection.assert_called_once()
|
|
self.connection.return_value.set_tunnel.assert_not_called()
|
|
self.assertEqual(self.connection.call_args.kwargs["context"].verify_mode, ssl.CERT_REQUIRED)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|