mirror of
https://framagit.org/framasoft/framaspace/argos.git
synced 2025-04-28 18:02:41 +02:00
109 lines
2.6 KiB
Python
109 lines
2.6 KiB
Python
from datetime import datetime, timedelta
|
|
|
|
import pytest
|
|
|
|
from argos.server import queries
|
|
from argos.server.models import Result, Task
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_old_results(db, ten_results):
|
|
assert db.query(Result).count() == 10
|
|
deleted = await queries.remove_old_results(db, 2)
|
|
assert deleted == 8
|
|
assert db.query(Result).count() == 2
|
|
# We should keep the last two results
|
|
assert db.query(Result).all() == ten_results[-2:]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_old_results_with_empty_db(db):
|
|
assert db.query(Result).count() == 0
|
|
deleted = await queries.remove_old_results(db, 2)
|
|
assert deleted == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_old_locks(db, ten_locked_tasks, ten_tasks):
|
|
assert db.query(Task).count() == 20
|
|
released = await queries.release_old_locks(db, 10)
|
|
assert released == 10
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_release_old_locks_with_empty_db(db):
|
|
assert db.query(Task).count() == 0
|
|
released = await queries.release_old_locks(db, 10)
|
|
assert released == 0
|
|
|
|
|
|
@pytest.fixture
|
|
def task(db):
|
|
task = Task(
|
|
url="https://www.example.com",
|
|
domain="example.com",
|
|
check="body-contains",
|
|
expected="foo",
|
|
frequency=1,
|
|
)
|
|
db.add(task)
|
|
db.commit()
|
|
return task
|
|
|
|
|
|
@pytest.fixture
|
|
def ten_results(db, task):
|
|
results = []
|
|
for i in range(10):
|
|
result = Result(
|
|
submitted_at=datetime.now(),
|
|
status="success",
|
|
context={"foo": "bar"},
|
|
task=task,
|
|
agent_id="test",
|
|
severity="ok",
|
|
)
|
|
db.add(result)
|
|
results.append(result)
|
|
db.commit()
|
|
return results
|
|
|
|
|
|
@pytest.fixture
|
|
def ten_locked_tasks(db):
|
|
a_minute_ago = datetime.now() - timedelta(minutes=1)
|
|
tasks = []
|
|
for i in range(10):
|
|
task = Task(
|
|
url="https://www.example.com",
|
|
domain="example.com",
|
|
check="body-contains",
|
|
expected="foo",
|
|
frequency=1,
|
|
selected_by="test",
|
|
selected_at=a_minute_ago,
|
|
)
|
|
db.add(task)
|
|
tasks.append(task)
|
|
db.commit()
|
|
return tasks
|
|
|
|
|
|
@pytest.fixture
|
|
def ten_tasks(db):
|
|
now = datetime.now()
|
|
tasks = []
|
|
for i in range(10):
|
|
task = Task(
|
|
url="https://www.example.com",
|
|
domain="example.com",
|
|
check="body-contains",
|
|
expected="foo",
|
|
frequency=1,
|
|
selected_by="test",
|
|
selected_at=now,
|
|
)
|
|
db.add(task)
|
|
tasks.append(task)
|
|
db.commit()
|
|
return tasks
|