mi
143 lines
5.1 KiB
Python
143 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from infratest import ConfigError, Reporter, Settings, as_bool, sanitize
|
|
|
|
|
|
def valid_env(ca_path: Path) -> str:
|
|
return f"""
|
|
HAN_PG_HOST=db.example.test
|
|
HAN_PG_PORT=6432
|
|
HAN_PG_DATABASE=han_chat
|
|
HAN_PG_SSLMODE=verify-full
|
|
HAN_PG_SSLROOTCERT={ca_path}
|
|
HAN_PG_PASSWORD_HAN_APP=han-password
|
|
HAN_PG_PASSWORD_BITRIX=bitrix-password
|
|
HAN_PG_PASSWORD_BITRIX_SYNC=sync-password
|
|
HAN_PG_PASSWORD_MESSAGE_SAFETY=safety-password
|
|
HAN_PG_PASSWORD_KEYCLOAK=keycloak-password
|
|
SELECTEL_S3_ENDPOINT_URL=https://s3.storage.selcloud.ru
|
|
SELECTEL_S3_BUCKET_QUARANTINE=quarantine
|
|
SELECTEL_S3_BUCKET_ATTACHMENTS=attachments
|
|
SELECTEL_S3_BUCKET_DOCUMENTS=documents
|
|
SELECTEL_S3_ACCESS_KEY=api-access
|
|
SELECTEL_S3_SECRET_KEY=api-secret
|
|
SELECTEL_S3_QUARANTINE_READ_ACCESS_KEY=read-access
|
|
SELECTEL_S3_QUARANTINE_READ_SECRET_KEY=read-secret
|
|
PUBLIC_WEB_URL=https://chat.example.test
|
|
INFRATEST_CHECK_CORS=true
|
|
INFRATEST_CHECK_MIGRATIONS=false
|
|
INFRATEST_TIMEOUT_SECONDS=10
|
|
""".strip()
|
|
|
|
|
|
class SettingsTests(unittest.TestCase):
|
|
def test_loads_complete_configuration_and_default_roles(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
ca = root / "ca.pem"
|
|
ca.write_text("test ca", encoding="utf-8")
|
|
env_file = root / ".env"
|
|
env_file.write_text(valid_env(ca), encoding="utf-8")
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
settings = Settings.from_env_file(env_file)
|
|
|
|
self.assertEqual(settings.pg_host, "db.example.test")
|
|
self.assertEqual(settings.pg_port, 6432)
|
|
self.assertEqual(
|
|
[(role.user, role.schema) for role in settings.pg_roles],
|
|
[
|
|
("han_app", "han_app"),
|
|
("bitrix_local_app", "bitrix_local"),
|
|
("bitrix_sync_user", "bitrix_sync"),
|
|
("message_safety_app", "message_safety"),
|
|
("keycloak_user", "keycloak"),
|
|
],
|
|
)
|
|
self.assertEqual(settings.buckets["documents"], "documents")
|
|
self.assertTrue(settings.check_cors)
|
|
self.assertFalse(settings.check_migrations)
|
|
|
|
def test_rejects_non_verifying_postgres_tls(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
ca = root / "ca.pem"
|
|
ca.write_text("test ca", encoding="utf-8")
|
|
env_file = root / ".env"
|
|
env_file.write_text(
|
|
valid_env(ca).replace(
|
|
"HAN_PG_SSLMODE=verify-full",
|
|
"HAN_PG_SSLMODE=require",
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(ConfigError, "verify-full"):
|
|
Settings.from_env_file(env_file)
|
|
|
|
def test_requires_public_url_when_cors_enabled(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
ca = root / "ca.pem"
|
|
ca.write_text("test ca", encoding="utf-8")
|
|
env_file = root / ".env"
|
|
env_file.write_text(
|
|
valid_env(ca).replace(
|
|
"PUBLIC_WEB_URL=https://chat.example.test",
|
|
"PUBLIC_WEB_URL=",
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaisesRegex(ConfigError, "PUBLIC_WEB_URL"):
|
|
Settings.from_env_file(env_file)
|
|
|
|
|
|
class SafetyTests(unittest.TestCase):
|
|
def test_sanitize_redacts_secret_and_url_query(self) -> None:
|
|
value = (
|
|
"password=do-not-print "
|
|
"url=https://bucket.example/object?X-Amz-Credential=secret&X-Amz-Signature=x"
|
|
)
|
|
sanitized = sanitize(value)
|
|
self.assertNotIn("do-not-print", sanitized)
|
|
self.assertNotIn("X-Amz", sanitized)
|
|
self.assertNotIn("Signature", sanitized)
|
|
self.assertIn("password=<redacted>", sanitized)
|
|
self.assertIn("https://bucket.example/object", sanitized)
|
|
|
|
def test_reporter_exit_code_and_json_are_secret_free(self) -> None:
|
|
reporter = Reporter()
|
|
with redirect_stdout(io.StringIO()):
|
|
reporter.add("ok", "PASS", "url=https://example.test/a?token=secret")
|
|
reporter.add("bad", "FAIL", "password=hidden")
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
report_path = Path(directory) / "report.json"
|
|
reporter.write_json(report_path)
|
|
data = json.loads(report_path.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(reporter.exit_code, 1)
|
|
serialized = json.dumps(data)
|
|
self.assertNotIn("hidden", serialized)
|
|
self.assertNotIn("?token", serialized)
|
|
|
|
def test_boolean_parser(self) -> None:
|
|
self.assertTrue(as_bool("yes"))
|
|
self.assertFalse(as_bool("OFF"))
|
|
with self.assertRaises(ConfigError):
|
|
as_bool("sometimes")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|