90 lines
3.9 KiB
Python
Executable File
90 lines
3.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""new-api 额度池离线预置工具定向回归测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
MODULE_PATH = Path(__file__).with_name("newapi_pool_provision.py")
|
|
SPEC = importlib.util.spec_from_file_location("newapi_pool_provision", MODULE_PATH)
|
|
assert SPEC and SPEC.loader
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
class ProvisionRequestTest(unittest.TestCase):
|
|
"""验证补建区间、宿主边界与敏感产物权限。"""
|
|
|
|
def test_builds_exact_neice_051_to_054_range(self) -> None:
|
|
names = MODULE.validate_provision_request(51, 4, "neice_", "mini-infra")
|
|
self.assertEqual(names, ["neice_051", "neice_052", "neice_053", "neice_054"])
|
|
|
|
def test_accepts_canonical_runtime_hostname(self) -> None:
|
|
names = MODULE.validate_provision_request(51, 4, "neice_", "minione-ubuntu-infra")
|
|
self.assertEqual(names, ["neice_051", "neice_052", "neice_053", "neice_054"])
|
|
|
|
def test_rejects_non_mini_infra_host(self) -> None:
|
|
with self.assertRaisesRegex(ValueError, "mini-infra"):
|
|
MODULE.validate_provision_request(51, 4, "neice_", "mini-desktop")
|
|
|
|
def test_rejects_empty_or_unsafe_ranges(self) -> None:
|
|
for start, count, prefix in ((0, 4, "neice_"), (51, 0, "neice_"), (51, 4, "../neice_")):
|
|
with self.subTest(start=start, count=count, prefix=prefix):
|
|
with self.assertRaises(ValueError):
|
|
MODULE.validate_provision_request(start, count, prefix, "mini-infra")
|
|
|
|
def test_emit_files_are_owner_read_write_only(self) -> None:
|
|
entry = {
|
|
"newapi_user_id": 51,
|
|
"newapi_token_id": 151,
|
|
"newapi_token_key": "sk-secret-value-must-stay-in-transport",
|
|
"grant_quota": 6849315,
|
|
"quota_per_unit_snapshot": 500000,
|
|
"usd_rate_snapshot": 7.3,
|
|
"status": "FREE",
|
|
"username": "neice_051",
|
|
}
|
|
with tempfile.TemporaryDirectory() as directory, mock.patch.object(MODULE, "datetime") as clock:
|
|
clock.now.return_value.strftime.return_value = "20260722-010203"
|
|
json_path, sql_path = MODULE.emit_outputs([entry], Path(directory), "neice_")
|
|
self.assertEqual(stat.S_IMODE(json_path.stat().st_mode), 0o600)
|
|
self.assertEqual(stat.S_IMODE(sql_path.stat().st_mode), 0o600)
|
|
self.assertIn("-- pool_entry username=neice_051 newapi_user_id=51", sql_path.read_text())
|
|
|
|
def test_waits_for_api_write_to_become_visible(self) -> None:
|
|
query = mock.Mock(side_effect=[None, None, 67])
|
|
with mock.patch.object(MODULE.time, "sleep") as sleep:
|
|
value = MODULE.wait_for_db_visibility(query, attempts=4, interval_seconds=0.1)
|
|
self.assertEqual(value, 67)
|
|
self.assertEqual(query.call_count, 3)
|
|
self.assertEqual(sleep.call_count, 2)
|
|
|
|
def test_db_visibility_wait_is_bounded(self) -> None:
|
|
query = mock.Mock(return_value=None)
|
|
with mock.patch.object(MODULE.time, "sleep") as sleep:
|
|
value = MODULE.wait_for_db_visibility(query, attempts=3, interval_seconds=0.1)
|
|
self.assertIsNone(value)
|
|
self.assertEqual(query.call_count, 3)
|
|
self.assertEqual(sleep.call_count, 2)
|
|
|
|
def test_postgres_attaches_sql_stdin_to_docker_exec(self) -> None:
|
|
completed = mock.Mock(returncode=0, stdout="67\n", stderr="")
|
|
with mock.patch.object(MODULE.subprocess, "run", return_value=completed) as run:
|
|
uid = MODULE.Postgres("secret").query_user_id("neice_051")
|
|
self.assertEqual(uid, 67)
|
|
command = run.call_args.args[0]
|
|
self.assertIn("-i", command)
|
|
self.assertEqual(run.call_args.kwargs["input"],
|
|
"SELECT id FROM users WHERE username='neice_051';")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|