136 lines
5.2 KiB
Python
136 lines
5.2 KiB
Python
import subprocess
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
class TransferServerRuntimeTests(unittest.TestCase):
|
|
def test_transfer_image_imports_server_module_from_app_workdir(self):
|
|
result = subprocess.run(
|
|
[
|
|
"docker", "run", "--rm",
|
|
"-e", "TRANSFER_CONFIG_PATH=/tmp/transfer_config.json",
|
|
"cloudsearch-transfer:test",
|
|
"python", "-c",
|
|
"import cloudsearch_transfer.server as s; print(hasattr(s, 'app'))",
|
|
],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("True", result.stdout)
|
|
|
|
def test_update_platform_creates_config_with_package_import(self):
|
|
result = subprocess.run(
|
|
[
|
|
"docker", "run", "--rm",
|
|
"-e", "TRANSFER_CONFIG_PATH=/tmp/transfer_config_server_test.json",
|
|
"cloudsearch-transfer:test",
|
|
"python", "-c",
|
|
"""
|
|
import os
|
|
os.environ['TRANSFER_API_TOKEN'] = 'test-token'
|
|
from cloudsearch_transfer.server import app
|
|
client = app.test_client()
|
|
resp = client.put('/api/config/platforms/newplatform', headers={'X-Transfer-Token': 'test-token'}, json={'enabled': True, 'save_dir': '/Media'})
|
|
print(resp.status_code)
|
|
print(resp.get_data(as_text=True))
|
|
""",
|
|
],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("200", result.stdout)
|
|
self.assertIn("newplatform", result.stdout)
|
|
|
|
def test_dockerfile_healthcheck_uses_http_health_probe(self):
|
|
dockerfile = Path(__file__).resolve().parents[1] / "Dockerfile"
|
|
content = dockerfile.read_text()
|
|
healthcheck_block = content.split("HEALTHCHECK", 1)[1].split("\n\nCMD", 1)[0]
|
|
self.assertIn("/health", healthcheck_block)
|
|
self.assertNotIn('python -m cloudsearch_transfer.server', healthcheck_block)
|
|
|
|
def test_config_update_requires_api_token_when_configured(self):
|
|
script = """
|
|
import os
|
|
os.environ['TRANSFER_API_TOKEN'] = 'secret-token'
|
|
os.environ['TRANSFER_CONFIG_PATH'] = '/tmp/transfer_config_auth_test.json'
|
|
from cloudsearch_transfer.server import app
|
|
client = app.test_client()
|
|
unauth = client.put('/api/config/platforms/newplatform', json={'enabled': True})
|
|
auth = client.put('/api/config/platforms/newplatform', headers={'X-Transfer-Token': 'secret-token'}, json={'enabled': True, 'save_dir': '/Media'})
|
|
print(unauth.status_code)
|
|
print(auth.status_code)
|
|
"""
|
|
result = subprocess.run(
|
|
[
|
|
"docker", "run", "--rm",
|
|
"-e", "TRANSFER_CONFIG_PATH=/tmp/transfer_config_auth_test.json",
|
|
"cloudsearch-transfer:test",
|
|
"python", "-c", script,
|
|
],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("401", result.stdout)
|
|
self.assertIn("200", result.stdout)
|
|
|
|
def test_config_update_requires_token_by_default(self):
|
|
script = """
|
|
import os
|
|
os.environ.pop('TRANSFER_API_TOKEN', None)
|
|
os.environ['TRANSFER_CONFIG_PATH'] = '/tmp/transfer_config_default_auth_test.json'
|
|
from cloudsearch_transfer.server import app
|
|
client = app.test_client()
|
|
resp = client.put('/api/config/platforms/newplatform', json={'enabled': True})
|
|
print(resp.status_code)
|
|
"""
|
|
result = subprocess.run(
|
|
[
|
|
"docker", "run", "--rm",
|
|
"-e", "TRANSFER_CONFIG_PATH=/tmp/transfer_config_default_auth_test.json",
|
|
"cloudsearch-transfer:test",
|
|
"python", "-c", script,
|
|
],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("401", result.stdout)
|
|
|
|
def test_config_save_persists_full_credentials_for_runtime_reload(self):
|
|
script = """
|
|
from cloudsearch_transfer.config import ConfigManager, PlatformConfig
|
|
path = '/tmp/transfer_config_persist_test.json'
|
|
cfg = ConfigManager(path)
|
|
cfg.platforms['quark'] = PlatformConfig(enabled=True, cookie='cookie-value-that-is-longer-than-twenty-characters', refresh_token='refresh-token-that-is-longer-than-twenty-characters')
|
|
cfg.save()
|
|
reloaded = ConfigManager(path)
|
|
print(reloaded.platforms['quark'].cookie)
|
|
print(reloaded.platforms['quark'].refresh_token)
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "run", "--rm", "cloudsearch-transfer:test", "python", "-c", script],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("cookie-value-that-is-longer-than-twenty-characters", result.stdout)
|
|
self.assertIn("refresh-token-that-is-longer-than-twenty-characters", result.stdout)
|
|
self.assertNotIn("...", result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|