118 lines
4.6 KiB
Python
118 lines
4.6 KiB
Python
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from _common import SETTINGS, agent
|
|
from fake_hub import FakeHub
|
|
from fake_relay import FakeRelay
|
|
|
|
|
|
class FolderTransportTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.root = os.path.join(self.tmp.name, "My Shared Files", "jira-bridge") # пробелы в пути
|
|
os.makedirs(self.root)
|
|
self.hub = FakeHub(SETTINGS)
|
|
self.relay = None
|
|
|
|
def tearDown(self):
|
|
if self.relay:
|
|
self.relay.stop()
|
|
self.hub.stop()
|
|
self.tmp.cleanup()
|
|
|
|
def client(self, timeout=5):
|
|
cfg = {"hub_url": "folder:" + self.root, "hub_token": self.hub.token, "timeout": timeout}
|
|
return agent.HubClient(cfg)
|
|
|
|
def run_relay(self, mode="forward", delay=0.0):
|
|
self.relay = FakeRelay(self.root, self.hub.base_url, mode, delay).start()
|
|
return self.relay
|
|
|
|
def listing(self, sub):
|
|
return sorted(os.listdir(os.path.join(self.root, sub)))
|
|
|
|
def test_settings_through_folder(self):
|
|
self.run_relay()
|
|
hub = self.client()
|
|
self.assertIsNotNone(hub.folder)
|
|
self.assertEqual(hub.settings()["project_key"], "AURORA")
|
|
self.assertEqual(self.listing("resp"), [])
|
|
self.assertEqual(self.listing("req"), [])
|
|
req = self.relay.requests[0]
|
|
self.assertEqual(req["method"], "GET")
|
|
self.assertEqual(req["path"], "/agent/settings")
|
|
self.assertEqual(req["authorization"], "Bearer " + self.hub.token)
|
|
self.assertEqual(len(req["id"]), 32)
|
|
|
|
def test_error_status_raises_http_error(self):
|
|
self.run_relay()
|
|
with self.assertRaises(agent.HttpError) as e:
|
|
self.client().result(999, {"status": "done"})
|
|
self.assertEqual(e.exception.code, 404)
|
|
|
|
def test_claim_deadline_at_least_45_seconds(self):
|
|
self.run_relay()
|
|
before = time.time()
|
|
self.client(timeout=5).claim(2, 1, None)
|
|
deadline = datetime.fromisoformat(self.relay.requests[0]["deadline"])
|
|
self.assertIsNotNone(deadline.tzinfo)
|
|
self.assertGreaterEqual(deadline.timestamp() - before, 44.5)
|
|
|
|
def test_no_answer_gives_connection_error_and_removes_request(self):
|
|
hub = self.client(timeout=0.5)
|
|
t0 = time.monotonic()
|
|
with self.assertRaises(ConnectionError) as e:
|
|
hub.settings()
|
|
self.assertLess(time.monotonic() - t0, 2)
|
|
self.assertIn("не ответил через папку", str(e.exception))
|
|
self.assertEqual([n for n in self.listing("req") if not n.startswith(".")], [])
|
|
|
|
def test_missing_folder_gives_connection_error(self):
|
|
cfg = {"hub_url": "folder:" + os.path.join(self.tmp.name, "нет"), "hub_token": "t", "timeout": 5}
|
|
with self.assertRaises(ConnectionError) as e:
|
|
agent.HubClient(cfg).settings()
|
|
self.assertIn("нет папки", str(e.exception))
|
|
|
|
def test_bad_json_response_gives_connection_error(self):
|
|
self.run_relay("garbage")
|
|
hub = self.client()
|
|
hub.folder.BAD_JSON_WAIT = 0.3
|
|
with self.assertRaises(ConnectionError) as e:
|
|
hub.settings()
|
|
self.assertIn("нечитаемый ответ", str(e.exception))
|
|
|
|
def test_late_response_is_not_reused(self):
|
|
self.run_relay("late", delay=1.0)
|
|
with self.assertRaises(ConnectionError):
|
|
self.client(timeout=0.3).settings()
|
|
time.sleep(1.3) # опоздавший ответ лёг в resp/
|
|
late = self.listing("resp")
|
|
self.assertEqual(len(late), 1)
|
|
self.relay.mode = "forward"
|
|
self.assertEqual(self.client().settings()["project_key"], "AURORA")
|
|
self.assertEqual(self.listing("resp"), late) # чужой (опоздавший) ответ не тронут
|
|
|
|
def test_foreign_files_untouched(self):
|
|
resp_dir = os.path.join(self.root, "resp")
|
|
os.makedirs(resp_dir, exist_ok=True)
|
|
foreign = os.path.join(resp_dir, "0" * 32 + ".json")
|
|
with open(foreign, "w", encoding="utf-8") as f:
|
|
json.dump({"id": "0" * 32, "status": 200, "body": "{}"}, f)
|
|
self.run_relay()
|
|
self.client().settings()
|
|
self.assertTrue(os.path.exists(foreign))
|
|
|
|
def test_http_url_keeps_http_transport(self):
|
|
cfg = {"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5}
|
|
hub = agent.HubClient(cfg)
|
|
self.assertIsNone(hub.folder)
|
|
self.assertEqual(hub.settings()["project_key"], "AURORA")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|