Spaces:
Running
Running
File size: 1,973 Bytes
6918275 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | import inspect
import threading
from flask import Flask, Response
from helpers import api, cache, extension, plugins, subagents
def test_real_profile_plugin_and_sync_startup_hook_are_discovered(monkeypatch):
from plugins.argus_supervisor.helpers import service
assert "argus-supervisor" in subagents.get_available_agents_dict(None)
assert "argus_supervisor" in plugins.get_enabled_plugins(None)
classes = extension._get_extension_classes("startup_migration", agent=None)
recovery = next(cls for cls in classes if cls.__name__ == "ArgusRecovery")
calls = []
monkeypatch.setattr(service, "kick", lambda **kwargs: calls.append(kwargs))
result = recovery(agent=None).execute()
assert not inspect.isawaitable(result) and calls == [{"force": True}]
assert any(
cls.__name__ == "ArgusHeartbeat"
for cls in extension._get_extension_classes("job_loop", agent=None)
)
def test_normal_plugin_api_loader_enforces_login_and_csrf(monkeypatch):
monkeypatch.setattr("helpers.login.get_credentials_hash", lambda: "hash")
app = Flask("argus_auth_contract")
app.secret_key = "unit-test-session"
app.add_url_rule("/", "serve_index", lambda: Response("index"))
app.add_url_rule("/login", "login_handler", lambda: Response("login"))
cache.clear(api.CACHE_AREA)
api.register_api_route(app, threading.RLock())
client = app.test_client()
url = "/api/plugins/argus_supervisor/health"
try:
assert client.post(url, json={}).status_code == 302
with client.session_transaction() as session:
session["authentication"] = "hash"
session["csrf_token"] = "csrf-unit"
assert client.post(url, json={}).status_code == 403
response = client.post(url, json={}, headers={"X-CSRF-Token": "csrf-unit"})
assert response.status_code == 200
assert "a2a_token_present" in response.get_json()
finally:
cache.clear(api.CACHE_AREA)
|