bellingcat/auto-archiver-api
1
0
Fork 0
mirror of https://github.com/bellingcat/auto-archiver-api.git synced 2026-06-12 21:48:35 +03:00
Code Issues Packages Projects Releases Wiki Activity

Format and lint the tests directory (#58)

Browse Source
This commit is contained in:
Michael Plunkett
2025-02-27 12:35:23 -06:00
committed by GitHub
parent 229db7dd5c
commit d575b6f9af
15 changed files with 1894 additions and 585 deletions
Download Patch File Download Diff File Expand all files Collapse all files

714
app/tests/web/db/test_crud.py
View File

@@ -2,125 +2,379 @@ from datetime import datetime, timedelta
from unittest.mock import patch
import pytest
import sqlalchemy
import yaml
from sqlalchemy import false, true
from sqlalchemy.sql import select
from app.shared.db import models
from app.shared.settings import Settings
from app.web.config import ALLOW_ANY_EMAIL
from app.web.db import crud
authors = ["rick@example.com", "morty@example.com", "jerry@example.com"]
@pytest.fixture()
def test_data(db_session):
# creates 3 users
for email in authors:
db_session.add(models.User(email=email))
db_session.commit()
assert db_session.query(models.User).count() == 3
# creates 100 archives for 3 users over 2 months with repeating URLs
for i in range(100):
author = authors[i % 3]
archive = models.Archive(
id=f"archive-id-456-{i}",
url=f"https://example-{i%3}.com",
result={},
public=author == "jerry@example.com",
author_id=author,
group_id="spaceship" if author == "morty@example.com" and i % 2 == 0 else None,
created_at=datetime(2021, (i % 2) + 1, (i % 25) + 1)
)
if i % 5 == 0:
archive.tags.append(models.Tag(id=f"tag-{i}"))
if i % 10 == 0:
archive.tags.append(models.Tag(id=f"tag-second-{i}"))
if i % 4 == 0:
archive.tags.append(models.Tag(id=f"tag-third-{i}"))
for j in range(10):
archive.urls.append(models.ArchiveUrl(url=f"https://example-{i}.com/{j}", key=f"media_{j}"))
db_session.add(archive)
# creates a sheet for each user
for i, email in enumerate(authors):
db_session.add(models.Sheet(id=f"sheet-{i}", name=f"sheet-{i}", author_id=email, group_id=None, frequency="daily"))
if email == "rick@example.com":
db_session.add(models.Sheet(id=f"sheet-{i}-2", name=f"sheet-{i}-2", author_id=email, group_id="spaceship", frequency="hourly"))
db_session.commit()
assert db_session.query(models.Archive).count() == 100
assert db_session.query(models.Tag).count() == 20 + 10 + 25
assert db_session.query(models.ArchiveUrl).count() == 1000
assert db_session.query(models.ArchiveUrl).filter(models.ArchiveUrl.archive_id == "archive-id-456-0").count() == 10
# setup groups
assert db_session.query(models.Group).count() == 0
from app.web.db import crud
crud.upsert_user_groups(db_session)
assert db_session.query(models.Group).count() == 4
assert db_session.query(models.User).count() == 3
def test_search_archives_by_url(test_data, db_session):
from app.web.config import ALLOW_ANY_EMAIL
# rick's archives are private
assert len(crud.search_archives_by_url(db_session, "https://example-0.com", "rick@example.com", True, False)) == 34
assert len(crud.search_archives_by_url(db_session, "https://example-0.com", "rick@example.com", [], False)) == 34
assert len(crud.search_archives_by_url(db_session, "https://example-0.com", "rick@example.com", [], True)) == 34
assert len(crud.search_archives_by_url(db_session, "https://example-0.com", ALLOW_ANY_EMAIL, [], False)) == 34
assert len(crud.search_archives_by_url(db_session, "https://example-0.com", ALLOW_ANY_EMAIL, True, False)) == 34
assert len(crud.search_archives_by_url(db_session, "https://example-0.com", "morty@example.com", [], False)) == 0
assert len(crud.search_archives_by_url(db_session, "https://example-0.com", "morty@example.com", [], True)) == 0
# Rick's archives are private
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-0.com",
"rick@example.com",
True,
False,
)
)
== 34
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-0.com",
"rick@example.com",
[],
False,
)
)
== 34
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-0.com",
"rick@example.com",
[],
True,
)
)
== 34
)
assert (
len(
crud.search_archives_by_url(
db_session, "https://example-0.com", ALLOW_ANY_EMAIL, [], False
)
)
== 34
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-0.com",
ALLOW_ANY_EMAIL,
True,
False,
)
)
== 34
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-0.com",
"morty@example.com",
[],
False,
)
)
== 0
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-0.com",
"morty@example.com",
[],
True,
)
)
== 0
)
# morty's archives are public but half are in spaceship group
assert len(crud.search_archives_by_url(db_session, "https://example-1.com", "rick@example.com", ["spaceship"], False)) == 16
assert len(crud.search_archives_by_url(db_session, "https://example-1.com", "rick@example.com", True, False)) == 16
assert len(crud.search_archives_by_url(db_session, "https://example-1.com", "jerry@example.com", True, True)) == 16
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-1.com",
"rick@example.com",
["spaceship"],
False,
)
)
== 16
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-1.com",
"rick@example.com",
True,
False,
)
)
== 16
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-1.com",
"jerry@example.com",
True,
True,
)
)
== 16
)
# jerry's archives are public
assert len(crud.search_archives_by_url(db_session, "https://example-2.com", "jerry@example.com", [], True)) == 33
assert len(crud.search_archives_by_url(db_session, "https://example-2.com", "rick@example.com", [], True)) == 33
# Jerry's archives are public
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-2.com",
"jerry@example.com",
[],
True,
)
)
== 33
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-2.com",
"rick@example.com",
[],
True,
)
)
== 33
)
# fuzzy search
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False)) == 100
assert len(crud.search_archives_by_url(db_session, "https://EXAMPLE", ALLOW_ANY_EMAIL, False, False)) == 100
assert len(crud.search_archives_by_url(db_session, "2.com", ALLOW_ANY_EMAIL, False, False)) == 33
assert (
len(
crud.search_archives_by_url(
db_session, "https://example", ALLOW_ANY_EMAIL, False, False
)
)
== 100
)
assert (
len(
crud.search_archives_by_url(
db_session, "https://EXAMPLE", ALLOW_ANY_EMAIL, False, False
)
)
== 100
)
assert (
len(
crud.search_archives_by_url(
db_session, "2.com", ALLOW_ANY_EMAIL, False, False
)
)
== 33
)
# absolute search
assert len(crud.search_archives_by_url(db_session, "example-2.com", ALLOW_ANY_EMAIL, [], False, absolute_search=True)) == 0
assert len(crud.search_archives_by_url(db_session, "https://example-2.com", ALLOW_ANY_EMAIL, [], False, absolute_search=True)) == 33
assert (
len(
crud.search_archives_by_url(
db_session,
"example-2.com",
ALLOW_ANY_EMAIL,
[],
False,
absolute_search=True,
)
)
== 0
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example-2.com",
ALLOW_ANY_EMAIL,
[],
False,
absolute_search=True,
)
)
== 33
)
# archived_after
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, True, True, archived_after=datetime(2010, 1, 1))) == 100
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, archived_after=datetime(2021, 1, 15))) == 70
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, archived_after=datetime(2031, 1, 1))) == 0
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
True,
True,
archived_after=datetime(2010, 1, 1),
)
)
== 100
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
archived_after=datetime(2021, 1, 15),
)
)
== 70
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
archived_after=datetime(2031, 1, 1),
)
)
== 0
)
# archived before
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, archived_before=datetime(2010, 1, 1))) == 0
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, archived_before=datetime(2021, 1, 15))) == 28
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, archived_before=datetime(2031, 1, 1))) == 100
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
archived_before=datetime(2010, 1, 1),
)
)
== 0
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
archived_before=datetime(2021, 1, 15),
)
)
== 28
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
archived_before=datetime(2031, 1, 1),
)
)
== 100
)
# archived before and after
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, archived_after=datetime(2001, 1, 1), archived_before=datetime(2031, 1, 11))) == 100
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, archived_after=datetime(2021, 1, 14), archived_before=datetime(2021, 1, 16))) == 2
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
archived_after=datetime(2001, 1, 1),
archived_before=datetime(2031, 1, 11),
)
)
== 100
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
archived_after=datetime(2021, 1, 14),
archived_before=datetime(2021, 1, 16),
)
)
== 2
)
# limit
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, limit=10)) == 10
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, limit=-1)) == 1
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
limit=10,
)
)
== 10
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
limit=-1,
)
)
== 1
)
# skip
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, False, False, skip=10)) == 90
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
False,
False,
skip=10,
)
)
== 90
)
def test_search_archives_by_email(test_data, db_session):
from app.web.config import ALLOW_ANY_EMAIL
# lower/upper case
assert len(crud.search_archives_by_email(db_session, "rick@example.com")) == 34
assert (
len(crud.search_archives_by_email(db_session, "rick@example.com")) == 34
)
# ALLOW_ANY_EMAIL is not a user
assert len(crud.search_archives_by_email(db_session, ALLOW_ANY_EMAIL)) == 0
@@ -138,45 +392,108 @@ def test_search_archives_by_email(test_data, db_session):
@patch("app.web.db.crud.DATABASE_QUERY_LIMIT", new=25)
def test_max_query_limit(test_data, db_session):
from app.web.config import ALLOW_ANY_EMAIL
assert (
len(
crud.search_archives_by_url(
db_session, "https://example", ALLOW_ANY_EMAIL, [], False
)
)
== 25
)
assert (
len(
crud.search_archives_by_url(
db_session,
"https://example",
ALLOW_ANY_EMAIL,
True,
True,
limit=1000,
)
)
== 25
)
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, [], False)) == 25
assert len(crud.search_archives_by_url(db_session, "https://example", ALLOW_ANY_EMAIL, True, True, limit=1000)) == 25
assert len(crud.search_archives_by_email(db_session, "rick@example.com")) == 25
assert len(crud.search_archives_by_email(db_session, "rick@example.com", limit=1000)) == 25
assert (
len(crud.search_archives_by_email(db_session, "rick@example.com")) == 25
)
assert (
len(
crud.search_archives_by_email(
db_session, "rick@example.com", limit=1000
)
)
== 25
)
def test_soft_delete(test_data, db_session):
# none deleted yet
db_session.query(models.Archive).filter(models.Archive.id == "archive-id-456-0").first() is not None
assert db_session.query(models.Archive).filter(models.Archive.deleted == True).count() == 0
assert (
db_session.query(models.Archive)
.filter(models.Archive.id == "archive-id-456-0")
.first()
is not None
)
assert (
db_session.query(models.Archive)
.filter(models.Archive.deleted.is_(true()))
.count()
== 0
)
# delete
assert crud.soft_delete_archive(db_session, "archive-id-456-0", "rick@example.com") == True
assert (
crud.soft_delete_archive(
db_session, "archive-id-456-0", "rick@example.com"
)
is True
)
# ensure soft delete
assert db_session.query(models.Archive).filter(models.Archive.deleted == True).count() == 1
db_session.query(models.Archive).filter(models.Archive.id == "archive-id-456-0").first() is None
assert (
db_session.query(models.Archive)
.filter(models.Archive.deleted.is_(true()))
.count()
== 1
)
assert (
db_session.query(models.Archive)
.filter(models.Archive.id == "archive-id-456-0")
.filter(models.Archive.deleted.is_(false()))
.first()
is None
)
# already deleted
assert crud.soft_delete_archive(db_session, "archive-id-456-0", "rick@example.com") == False
assert (
crud.soft_delete_archive(
db_session, "archive-id-456-0", "rick@example.com"
)
is False
)
def test_count_archives(test_data, db_session):
assert crud.count_archives(db_session) == 100
db_session.query(models.Archive).filter(models.Archive.id == "archive-id-456-0").delete()
db_session.query(models.Archive).filter(
models.Archive.id == "archive-id-456-0"
).delete()
db_session.commit()
assert crud.count_archives(db_session) == 99
def test_count_archive_urls(test_data, db_session):
assert crud.count_archive_urls(db_session) == 1000
db_session.query(models.ArchiveUrl).filter(models.ArchiveUrl.url == "https://example-0.com/0").delete()
db_session.query(models.ArchiveUrl).filter(
models.ArchiveUrl.url == "https://example-0.com/0"
).delete()
db_session.commit()
assert crud.count_archive_urls(db_session) == 999
db_session.query(models.Archive).filter(models.Archive.id == "archive-id-456-0").delete()
db_session.query(models.Archive).filter(
models.Archive.id == "archive-id-456-0"
).delete()
db_session.commit()
# no Cascade is enabled
assert crud.count_archives(db_session) == 99
@@ -185,16 +502,23 @@ def test_count_archive_urls(test_data, db_session):
def test_count_users(test_data, db_session):
assert crud.count_users(db_session) == 3
db_session.query(models.User).filter(models.User.email == "rick@example.com").delete()
db_session.query(models.User).filter(
models.User.email == "rick@example.com"
).delete()
db_session.commit()
assert crud.count_users(db_session) == 2
def test_count_by_users_since(test_data, db_session):
from app.web.db import crud
# 100y window
assert len(cu := crud.count_by_user_since(db_session, 60 * 60 * 24 * 31 * 12 * 100)) == 3
assert (
len(
cu := crud.count_by_user_since(
db_session, 60 * 60 * 24 * 31 * 12 * 100
)
)
== 3
)
assert cu[0].total == 34
assert cu[1].total == 33
assert cu[2].total == 33
@@ -203,9 +527,18 @@ def test_count_by_users_since(test_data, db_session):
def test_upsert_group(test_data, db_session):
assert db_session.query(models.Group).count() == 4
repeatable_params = ["desc 1", "orch.yaml", "sheet.yaml", "service_account_email@example.com", {"read": ["all"]}, ["example.com"]]
repeatable_params = [
"desc 1",
"orch.yaml",
"sheet.yaml",
"service_account_email@example.com",
{"read": ["all"]},
["example.com"],
]
assert (g1 := crud.upsert_group(db_session, "spaceship", *repeatable_params)) is not None
assert (
g1 := crud.upsert_group(db_session, "spaceship", *repeatable_params)
) is not None
assert g1.id == "spaceship"
assert g1.description == "desc 1"
assert g1.orchestrator == "orch.yaml"
@@ -214,14 +547,25 @@ def test_upsert_group(test_data, db_session):
assert g1.permissions == {"read": ["all"]}
assert g1.domains == ["example.com"]
assert len(g1.users) == 2
assert [u.email for u in g1.users] == ["rick@example.com", "morty@example.com"]
assert [u.email for u in g1.users] == [
"rick@example.com",
"morty@example.com",
]
assert (g2 := crud.upsert_group(db_session, "interdimensional", *repeatable_params)) is not None
assert (
g2 := crud.upsert_group(
db_session, "interdimensional", *repeatable_params
)
) is not None
assert g2.id == "interdimensional"
assert len(g2.users) == 1
assert [u.email for u in g2.users] == ["rick@example.com"]
assert (g3 := crud.upsert_group(db_session, "this-is-a-new-group", *repeatable_params)) is not None
assert (
g3 := crud.upsert_group(
db_session, "this-is-a-new-group", *repeatable_params
)
) is not None
assert g3.id == "this-is-a-new-group"
assert len(g3.users) == 0
@@ -229,29 +573,38 @@ def test_upsert_group(test_data, db_session):
def test_upsert_user_groups(db_session):
@patch('app.web.db.crud.get_settings', new=lambda: bad_setings)
@patch("app.web.db.crud.get_settings", new=lambda: bad_settings)
def test_missing_yaml(db_session):
with pytest.raises(FileNotFoundError):
crud.upsert_user_groups(db_session)
@patch('app.web.db.crud.get_settings', new=lambda: bad_setings)
@patch("app.web.db.crud.get_settings", new=lambda: bad_settings)
def test_broken_yaml(db_session):
with pytest.raises(yaml.YAMLError):
crud.upsert_user_groups(db_session)
bad_setings = Settings(_env_file=".env.test")
bad_settings = Settings(_env_file=".env.test")
bad_setings.USER_GROUPS_FILENAME = "app/tests/user-groups.test.missing.yaml"
bad_settings.USER_GROUPS_FILENAME = (
"app/tests/user-groups.test.missing.yaml"
)
test_missing_yaml(db_session)
bad_setings.USER_GROUPS_FILENAME = "app/tests/user-groups.test.broken.yaml"
bad_settings.USER_GROUPS_FILENAME = "app/tests/user-groups.test.broken.yaml"
test_broken_yaml(db_session)
def test_create_sheet(db_session):
assert db_session.query(models.Sheet).count() == 0
s = crud.create_sheet(db_session, "sheet-id-123", "sheet name", "email@example.com", "group-id", "hourly")
s = crud.create_sheet(
db_session,
"sheet-id-123",
"sheet name",
"email@example.com",
"group-id",
"hourly",
)
assert s is not None
assert s.id == "sheet-id-123"
assert s.name == "sheet name"
@@ -261,19 +614,35 @@ def test_create_sheet(db_session):
assert db_session.query(models.Sheet).count() == 1
# duplicate id
import sqlalchemy
with pytest.raises(sqlalchemy.exc.IntegrityError):
crud.create_sheet(db_session, "sheet-id-123", "I thought this was another sheet", "email", "group-id", "hourly")
crud.create_sheet(
db_session,
"sheet-id-123",
"I thought this was another sheet",
"email",
"group-id",
"hourly",
)
def test_get_user_sheet(test_data, db_session):
assert crud.get_user_sheet(db_session, "", "sheet-0") is None
assert crud.get_user_sheet(db_session, "morty@example.com", "sheet-0") is None
assert (
crud.get_user_sheet(db_session, "morty@example.com", "sheet-0") is None
)
assert crud.get_user_sheet(db_session, "rick@example.com", "sheet-0") is not None
assert crud.get_user_sheet(db_session, "rick@example.com", "sheet-0-2") is not None
assert crud.get_user_sheet(db_session, "morty@example.com", "sheet-1") is not None
assert (
crud.get_user_sheet(db_session, "rick@example.com", "sheet-0")
is not None
)
assert (
crud.get_user_sheet(db_session, "rick@example.com", "sheet-0-2")
is not None
)
assert (
crud.get_user_sheet(db_session, "morty@example.com", "sheet-1")
is not None
)
def test_get_user_sheets(test_data, db_session):
@@ -285,9 +654,9 @@ def test_get_user_sheets(test_data, db_session):
def test_delete_sheet(test_data, db_session):
assert crud.delete_sheet(db_session, "sheet-0", "") == False
assert crud.delete_sheet(db_session, "sheet-0", "rick@example.com") == True
assert crud.delete_sheet(db_session, "sheet-0", "rick@example.com") == False
assert crud.delete_sheet(db_session, "sheet-0", "") is False
assert crud.delete_sheet(db_session, "sheet-0", "rick@example.com") is True
assert crud.delete_sheet(db_session, "sheet-0", "rick@example.com") is False
@pytest.mark.asyncio
@@ -299,21 +668,21 @@ async def test_find_by_store_until(async_db_session):
url="https://example-expired-1.com",
result={},
author_id="rick@example.com",
store_until=now - timedelta(days=1)
store_until=now - timedelta(days=1),
)
archive2 = models.Archive(
id="archive-expired-2",
url="https://example-expired-2.com",
result={},
author_id="rick@example.com",
store_until=now - timedelta(hours=1)
store_until=now - timedelta(hours=1),
)
archive3 = models.Archive(
id="archive-active",
url="https://example-active.com",
result={},
author_id="rick@example.com",
store_until=now + timedelta(days=1)
store_until=now + timedelta(days=1),
)
async_db_session.add_all([archive1, archive2, archive3])
await async_db_session.commit()
@@ -323,11 +692,15 @@ async def test_find_by_store_until(async_db_session):
assert len(list(expired)) == 2
# Should find 1 archive expired before 2 hours ago
expired = await crud.find_by_store_until(async_db_session, now - timedelta(hours=2))
expired = await crud.find_by_store_until(
async_db_session, now - timedelta(hours=2)
)
assert len(list(expired)) == 1
# Should find no archives expired before 2 days ago
expired = await crud.find_by_store_until(async_db_session, now - timedelta(days=2))
expired = await crud.find_by_store_until(
async_db_session, now - timedelta(days=2)
)
assert len(list(expired)) == 0
# Should not find deleted archives
@@ -339,45 +712,78 @@ async def test_find_by_store_until(async_db_session):
@pytest.mark.asyncio
async def test_get_sheets_by_id_hash(async_db_session):
author_emails = ["rick@example.com", "morty@example.com", "jerry@example.com"]
# Add test data
authors = ["rick@example.com", "morty@example.com", "jerry@example.com"]
sheets = [
models.Sheet(id="sheet-0", name="sheet-0", author_id=authors[0], group_id=None, frequency="daily"),
models.Sheet(id="sheet-0-2", name="sheet-0-2", author_id=authors[0], group_id="spaceship", frequency="hourly"),
models.Sheet(id="sheet-1", name="sheet-1", author_id=authors[1], group_id=None, frequency="daily"),
models.Sheet(id="sheet-2", name="sheet-2", author_id=authors[2], group_id=None, frequency="daily")
models.Sheet(
id="sheet-0",
name="sheet-0",
author_id=author_emails[0],
group_id=None,
frequency="daily",
),
models.Sheet(
id="sheet-0-2",
name="sheet-0-2",
author_id=author_emails[0],
group_id="spaceship",
frequency="hourly",
),
models.Sheet(
id="sheet-1",
name="sheet-1",
author_id=author_emails[1],
group_id=None,
frequency="daily",
),
models.Sheet(
id="sheet-2",
name="sheet-2",
author_id=author_emails[2],
group_id=None,
frequency="daily",
),
]
async_db_session.add_all(sheets)
await async_db_session.commit()
with patch("app.web.db.crud.fnv1a_hash_mod", return_value=1):
# Test retrieving hourly sheets
hourly_sheets = await crud.get_sheets_by_id_hash(async_db_session, "hourly", 4, 1)
hourly_sheets = await crud.get_sheets_by_id_hash(
async_db_session, "hourly", 4, 1
)
assert len(hourly_sheets) == 1
assert hourly_sheets[0].id == "sheet-0-2"
assert hourly_sheets[0].frequency == "hourly"
# Test retrieving daily sheets
daily_sheets = await crud.get_sheets_by_id_hash(async_db_session, "daily", 4, 1)
daily_sheets = await crud.get_sheets_by_id_hash(
async_db_session, "daily", 4, 1
)
assert len(daily_sheets) == 3
assert all(sheet.frequency == "daily" for sheet in daily_sheets)
assert {sheet.id for sheet in daily_sheets} == {"sheet-0", "sheet-1", "sheet-2"}
assert {sheet.id for sheet in daily_sheets} == {
"sheet-0",
"sheet-1",
"sheet-2",
}
# Test with non-matching hash
no_sheets = await crud.get_sheets_by_id_hash(async_db_session, "daily", 4, 3)
no_sheets = await crud.get_sheets_by_id_hash(
async_db_session, "daily", 4, 3
)
assert len(no_sheets) == 0
# Test with non-existent frequency
weekly_sheets = await crud.get_sheets_by_id_hash(async_db_session, "weekly", 4, 1)
weekly_sheets = await crud.get_sheets_by_id_hash(
async_db_session, "weekly", 4, 1
)
assert len(weekly_sheets) == 0
@pytest.mark.asyncio
async def test_delete_stale_sheets(async_db_session):
from datetime import datetime, timedelta
from sqlalchemy.sql import select
now = datetime.now()
active_date = now - timedelta(days=5)
stale_date = now - timedelta(days=15)
@@ -389,29 +795,29 @@ async def test_delete_stale_sheets(async_db_session):
name="Active Sheet 1",
author_id="rick@example.com",
frequency="daily",
last_url_archived_at=active_date
last_url_archived_at=active_date,
),
models.Sheet(
id="sheet-active-2",
name="Active Sheet 2",
author_id="morty@example.com",
frequency="hourly",
last_url_archived_at=active_date
last_url_archived_at=active_date,
),
models.Sheet(
id="sheet-stale-1",
name="Stale Sheet 1",
author_id="rick@example.com",
frequency="daily",
last_url_archived_at=stale_date
last_url_archived_at=stale_date,
),
models.Sheet(
id="sheet-stale-2",
name="Stale Sheet 2",
author_id="morty@example.com",
frequency="daily",
last_url_archived_at=stale_date
)
last_url_archived_at=stale_date,
),
]
async_db_session.add_all(sheets)
await async_db_session.commit()

613
app/tests/web/db/test_user_state.py
View File

@@ -1,4 +1,3 @@
from unittest.mock import MagicMock, PropertyMock, patch
import pytest
@@ -6,6 +5,7 @@ import pytest
from app.shared.db import models
from app.shared.user_groups import GroupInfo, GroupPermissions
from app.web.db.user_state import UserState
from app.web.utils.misc import convert_priority_to_queue_dict
def fresh_user_state():
@@ -21,39 +21,73 @@ def user_state():
def user_state_with_groups(user_state):
user_groups = [
models.Group(id="no-permissions", permissions={}),
models.Group(id="group1", description="this is g1", service_account_email="sa1@example.com", permissions={"read": ["group1", "no-permissions"], "read_public": True, "archive_url": True, "archive_sheet": True, "max_archive_lifespan_months": 24, "max_monthly_urls": 100, "max_monthly_mbs": 1000, "priority": "high"}),
models.Group(id="group2", description="this is g2", service_account_email="sa2@example.com", permissions={"read": ["all"], "read_public": True, "archive_url": False, "archive_sheet": False, "max_archive_lifespan_months": -1, "max_monthly_urls": -1, "max_monthly_mbs": -1, "priority": "low", "sheet_frequency": {"daily"}}),
models.Group(
id="group1",
description="this is g1",
service_account_email="sa1@example.com",
permissions={
"read": ["group1", "no-permissions"],
"read_public": True,
"archive_url": True,
"archive_sheet": True,
"max_archive_lifespan_months": 24,
"max_monthly_urls": 100,
"max_monthly_mbs": 1000,
"priority": "high",
},
),
models.Group(
id="group2",
description="this is g2",
service_account_email="sa2@example.com",
permissions={
"read": ["all"],
"read_public": True,
"archive_url": False,
"archive_sheet": False,
"max_archive_lifespan_months": -1,
"max_monthly_urls": -1,
"max_monthly_mbs": -1,
"priority": "low",
"sheet_frequency": {"daily"},
},
),
]
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=user_groups):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=user_groups,
):
yield user_state
def test_permissions(user_state_with_groups):
permissions = user_state_with_groups.permissions
assert permissions["all"].read == True
assert permissions["all"].read_public == True
assert permissions["all"].archive_url == True
assert permissions["all"].archive_sheet == True
assert permissions["all"].read is True
assert permissions["all"].read_public is True
assert permissions["all"].archive_url is True
assert permissions["all"].archive_sheet is True
assert permissions["all"].max_archive_lifespan_months == -1
assert permissions["all"].max_monthly_urls == -1
assert permissions["all"].max_monthly_mbs == -1
assert permissions["all"].priority == "high"
assert permissions["group1"].read == set(["group1", "no-permissions"])
assert permissions["group1"].read_public == True
assert permissions["group1"].archive_url == True
assert permissions["group1"].archive_sheet == True
assert permissions["group1"].read == {"group1", "no-permissions"}
assert permissions["group1"].read_public is True
assert permissions["group1"].archive_url is True
assert permissions["group1"].archive_sheet is True
assert permissions["group1"].max_archive_lifespan_months == 24
assert permissions["group1"].max_monthly_urls == 100
assert permissions["group1"].max_monthly_mbs == 1000
assert permissions["group1"].priority == "high"
assert permissions["group2"].read == set(["all"])
assert permissions["group2"].read_public == True
assert permissions["group2"].archive_url == False
assert permissions["group2"].archive_sheet == False
assert permissions["group2"].read == {"all"}
assert permissions["group2"].read_public is True
assert permissions["group2"].archive_url is False
assert permissions["group2"].archive_sheet is False
assert permissions["group2"].max_archive_lifespan_months == -1
assert permissions["group2"].max_monthly_urls == -1
assert permissions["group2"].max_monthly_mbs == -1
@@ -63,13 +97,19 @@ def test_permissions(user_state_with_groups):
def test_user_groups_names(user_state):
with patch('app.web.db.crud.get_user_group_names', return_value=["group1", "group2"]) as mock:
with patch(
"app.web.db.crud.get_user_group_names",
return_value=["group1", "group2"],
) as mock:
assert user_state.user_groups_names == ["group1", "group2", "default"]
mock.assert_called_once_with(None, "test@example.com")
def test_user_groups(user_state):
with patch('app.web.db.crud.get_user_groups_by_name', return_value=[MagicMock(), MagicMock()]) as mock:
with patch(
"app.web.db.crud.get_user_groups_by_name",
return_value=[MagicMock(), MagicMock()],
) as mock:
user_state._user_groups_names = ["group1", "group2"]
assert len(user_state.user_groups) == 2
mock.assert_called_once_with(None, ["group1", "group2"])
@@ -78,85 +118,166 @@ def test_user_groups(user_state):
def test_read():
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(us, "_read")
assert us.read == set()
assert us._read == set()
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"read": ["group1", "no-permissions"]})]):
assert us.read == set(["group1", "no-permissions"])
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(
id="group1", permissions={"read": ["group1", "no-permissions"]}
)
],
):
assert us.read == {"group1", "no-permissions"}
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"read": ["all"]})]):
assert us.read == True
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="group1", permissions={"read": ["all"]})],
):
assert us.read is True
def test_read_public():
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(us, "_read_public")
assert us.read_public == False
assert us._read_public == False
assert us.read_public is False
assert us._read_public is False
mock.assert_called_once()
# no new calls
assert us.read_public == False
assert us.read_public is False
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"read_public": True})]):
assert us.read_public == True
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"read_public": True})
],
):
assert us.read_public is True
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"read_public": False})]):
assert us.read_public == False
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"read_public": False})
],
):
assert us.read_public is False
def test_archive_url():
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(us, "_archive_url")
assert us.archive_url == False
assert us._archive_url == False
assert us.archive_url is False
assert us._archive_url is False
mock.assert_called_once()
# no new calls
assert us.archive_url == False
assert us.archive_url is False
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"archive_url": False})]):
assert us.archive_url == False
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"archive_url": False})
],
):
assert us.archive_url is False
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"archive_url": True})]):
assert us.archive_url == True
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"archive_url": True})
],
):
assert us.archive_url is True
def test_archive_sheet():
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(us, "_archive_sheet")
assert us.archive_sheet == False
assert us._archive_sheet == False
assert us.archive_sheet is False
assert us._archive_sheet is False
mock.assert_called_once()
# no new calls
assert us.archive_sheet == False
assert us.archive_sheet is False
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"archive_sheet": False})]):
assert us.archive_sheet == False
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"archive_sheet": False})
],
):
assert us.archive_sheet is False
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"archive_sheet": True})]):
assert us.archive_sheet == True
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"archive_sheet": True})
],
):
assert us.archive_sheet is True
def test_sheet_frequency():
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(us, "_sheet_frequency")
assert us.sheet_frequency == set()
assert us._sheet_frequency == set()
@@ -166,18 +287,42 @@ def test_sheet_frequency():
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"sheet_frequency": ["daily", "hourly"]})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(
id="group1",
permissions={"sheet_frequency": ["daily", "hourly"]},
)
],
):
assert us.sheet_frequency == {"daily", "hourly"}
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"sheet_frequency": []})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"sheet_frequency": []})
],
):
assert us.sheet_frequency == set()
def test_max_archive_lifespan_months():
us = fresh_user_state()
default = GroupPermissions.model_fields["max_archive_lifespan_months"].default
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
default = GroupPermissions.model_fields[
"max_archive_lifespan_months"
].default
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(us, "_max_archive_lifespan_months")
assert us.max_archive_lifespan_months == default
assert us._max_archive_lifespan_months == default
@@ -187,18 +332,44 @@ def test_max_archive_lifespan_months():
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"max_archive_lifespan_months": 24})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(
id="group1", permissions={"max_archive_lifespan_months": 24}
)
],
):
assert us.max_archive_lifespan_months == 24
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"max_archive_lifespan_months": 150}), models.Group(id="group2", permissions={"max_archive_lifespan_months": -1})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(
id="group1", permissions={"max_archive_lifespan_months": 150}
),
models.Group(
id="group2", permissions={"max_archive_lifespan_months": -1}
),
],
):
assert us.max_archive_lifespan_months == -1
def test_max_monthly_urls():
us = fresh_user_state()
default = GroupPermissions.model_fields["max_monthly_urls"].default
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(us, "_max_monthly_urls")
assert us.max_monthly_urls == default
assert us._max_monthly_urls == default
@@ -208,18 +379,38 @@ def test_max_monthly_urls():
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"max_monthly_urls": 100})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"max_monthly_urls": 100})
],
):
assert us.max_monthly_urls == 100
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"max_monthly_urls": 150}), models.Group(id="group2", permissions={"max_monthly_urls": -1})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"max_monthly_urls": 150}),
models.Group(id="group2", permissions={"max_monthly_urls": -1}),
],
):
assert us.max_monthly_urls == -1
def test_max_monthly_mbs():
us = fresh_user_state()
default = GroupPermissions.model_fields["max_monthly_mbs"].default
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(us, "_max_monthly_mbs")
assert us.max_monthly_mbs == default
assert us._max_monthly_mbs == default
@@ -229,17 +420,37 @@ def test_max_monthly_mbs():
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"max_monthly_mbs": 1000})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"max_monthly_mbs": 1000})
],
):
assert us.max_monthly_mbs == 1000
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"max_monthly_mbs": 1500}), models.Group(id="group2", permissions={"max_monthly_mbs": -1})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"max_monthly_mbs": 1500}),
models.Group(id="group2", permissions={"max_monthly_mbs": -1}),
],
):
assert us.max_monthly_mbs == -1
def test_priority(user_state):
default = GroupPermissions.model_fields["priority"].default
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="no-permissions", permissions={})]) as mock:
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[models.Group(id="no-permissions", permissions={})],
) as mock:
assert not hasattr(user_state, "_priority")
assert user_state.priority == default
assert user_state._priority == default
@@ -249,11 +460,26 @@ def test_priority(user_state):
mock.assert_called_once()
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"priority": "high"})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"priority": "high"})
],
):
assert us.priority == "high"
us = fresh_user_state()
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[models.Group(id="group1", permissions={"priority": "low"}), models.Group(id="group2", permissions={"priority": "medium"})]):
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"priority": "low"}),
models.Group(id="group2", permissions={"priority": "medium"}),
],
):
assert us.priority == "low"
@@ -263,21 +489,45 @@ def test_active():
(True, False, False, False, True),
(False, True, False, False, True),
(False, False, True, False, True),
(False, False, False, True, True)
(False, False, False, True, True),
]:
us = fresh_user_state()
with patch.object(UserState, 'read', new_callable=PropertyMock, return_value=read), \
patch.object(UserState, 'read_public', new_callable=PropertyMock, return_value=read_public), \
patch.object(UserState, 'archive_url', new_callable=PropertyMock, return_value=archive_url), \
patch.object(UserState, 'archive_sheet', new_callable=PropertyMock, return_value=archive_sheet):
with (
patch.object(
UserState, "read", new_callable=PropertyMock, return_value=read
),
patch.object(
UserState,
"read_public",
new_callable=PropertyMock,
return_value=read_public,
),
patch.object(
UserState,
"archive_url",
new_callable=PropertyMock,
return_value=archive_url,
),
patch.object(
UserState,
"archive_sheet",
new_callable=PropertyMock,
return_value=archive_sheet,
),
):
assert us.active == is_active
def test_in_group(user_state):
with patch.object(UserState, 'user_groups_names', new_callable=PropertyMock, return_value=["group1", "group2"]):
assert user_state.in_group("group1") == True
assert user_state.in_group("group2") == True
assert user_state.in_group("group3") == False
with patch.object(
UserState,
"user_groups_names",
new_callable=PropertyMock,
return_value=["group1", "group2"],
):
assert user_state.in_group("group1") is True
assert user_state.in_group("group2") is True
assert user_state.in_group("group3") is False
def test_usage(db_session):
@@ -295,10 +545,34 @@ def test_usage(db_session):
]
megabytes = int(sum(bytes) / 1024 / 1024)
with patch.object(db_session, 'query', side_effect=[
MagicMock(filter=MagicMock(return_value=MagicMock(group_by=MagicMock(return_value=MagicMock(all=MagicMock(return_value=user_sheets)))))),
MagicMock(filter=MagicMock(return_value=MagicMock(group_by=MagicMock(return_value=MagicMock(all=MagicMock(return_value=urls_by_group))))))
]):
with patch.object(
db_session,
"query",
side_effect=[
MagicMock(
filter=MagicMock(
return_value=MagicMock(
group_by=MagicMock(
return_value=MagicMock(
all=MagicMock(return_value=user_sheets)
)
)
)
)
),
MagicMock(
filter=MagicMock(
return_value=MagicMock(
group_by=MagicMock(
return_value=MagicMock(
all=MagicMock(return_value=urls_by_group)
)
)
)
)
),
],
):
usage_response = user_state.usage()
assert usage_response.monthly_urls == 155
@@ -306,11 +580,15 @@ def test_usage(db_session):
assert usage_response.total_sheets == 115
assert usage_response.groups["group1"].monthly_urls == 50
assert usage_response.groups["group1"].monthly_mbs == int(bytes[0] / 1024 / 1024)
assert usage_response.groups["group1"].monthly_mbs == int(
bytes[0] / 1024 / 1024
)
assert usage_response.groups["group1"].total_sheets == 5
assert usage_response.groups["group2"].monthly_urls == 100
assert usage_response.groups["group2"].monthly_mbs == int(bytes[1] / 1024 / 1024)
assert usage_response.groups["group2"].monthly_mbs == int(
bytes[1] / 1024 / 1024
)
assert usage_response.groups["group2"].total_sheets == 10
assert usage_response.groups["group3"].monthly_urls == 0
@@ -318,7 +596,9 @@ def test_usage(db_session):
assert usage_response.groups["group3"].total_sheets == 100
assert usage_response.groups["group4"].monthly_urls == 5
assert usage_response.groups["group4"].monthly_mbs == int(bytes[2] / 1024 / 1024)
assert usage_response.groups["group4"].monthly_mbs == int(
bytes[2] / 1024 / 1024
)
assert usage_response.groups["group4"].total_sheets == 0
@@ -334,8 +614,23 @@ def test_has_quota_monthly_sheets(db_session):
]
for permissions, count, expected in test_cases:
with patch.object(UserState, 'permissions', new_callable=PropertyMock, return_value=permissions):
with patch.object(us.db, 'query', return_value=MagicMock(filter=MagicMock(return_value=MagicMock(count=MagicMock(return_value=count))))):
with patch.object(
UserState,
"permissions",
new_callable=PropertyMock,
return_value=permissions,
):
with patch.object(
us.db,
"query",
return_value=MagicMock(
filter=MagicMock(
return_value=MagicMock(
count=MagicMock(return_value=count)
)
)
),
):
assert us.has_quota_monthly_sheets("group1") == expected
@@ -350,8 +645,23 @@ def test_has_quota_max_monthly_urls(db_session):
]
for permissions, count, expected in test_cases:
with patch.object(UserState, 'permissions', new_callable=PropertyMock, return_value=permissions):
with patch.object(us.db, 'query', return_value=MagicMock(filter=MagicMock(return_value=MagicMock(count=MagicMock(return_value=count))))):
with patch.object(
UserState,
"permissions",
new_callable=PropertyMock,
return_value=permissions,
):
with patch.object(
us.db,
"query",
return_value=MagicMock(
filter=MagicMock(
return_value=MagicMock(
count=MagicMock(return_value=count)
)
)
),
):
assert us.has_quota_max_monthly_urls("group1") == expected
test_cases = [
(-1, 1000, True),
@@ -361,8 +671,23 @@ def test_has_quota_max_monthly_urls(db_session):
]
for max_urls, count, expected in test_cases:
with patch.object(UserState, 'max_monthly_urls', new_callable=PropertyMock, return_value=max_urls):
with patch.object(us.db, 'query', return_value=MagicMock(filter=MagicMock(return_value=MagicMock(count=MagicMock(return_value=count))))):
with patch.object(
UserState,
"max_monthly_urls",
new_callable=PropertyMock,
return_value=max_urls,
):
with patch.object(
us.db,
"query",
return_value=MagicMock(
filter=MagicMock(
return_value=MagicMock(
count=MagicMock(return_value=count)
)
)
),
):
assert us.has_quota_max_monthly_urls("") == expected
@@ -377,8 +702,29 @@ def test_has_quota_max_monthly_mbs(db_session):
]
for permissions, mbs, expected in test_cases:
with patch.object(UserState, 'permissions', new_callable=PropertyMock, return_value=permissions):
with patch.object(us.db, 'query', return_value=MagicMock(filter=MagicMock(return_value=MagicMock(with_entities=MagicMock(return_value=MagicMock(scalar=MagicMock(return_value=mbs * 1024 * 1024))))))):
with patch.object(
UserState,
"permissions",
new_callable=PropertyMock,
return_value=permissions,
):
with patch.object(
us.db,
"query",
return_value=MagicMock(
filter=MagicMock(
return_value=MagicMock(
with_entities=MagicMock(
return_value=MagicMock(
scalar=MagicMock(
return_value=mbs * 1024 * 1024
)
)
)
)
)
),
):
assert us.has_quota_max_monthly_mbs("group1") == expected
test_cases = [
@@ -389,8 +735,29 @@ def test_has_quota_max_monthly_mbs(db_session):
]
for max_mbs, mbs, expected in test_cases:
with patch.object(UserState, 'max_monthly_mbs', new_callable=PropertyMock, return_value=max_mbs):
with patch.object(us.db, 'query', return_value=MagicMock(filter=MagicMock(return_value=MagicMock(with_entities=MagicMock(return_value=MagicMock(scalar=MagicMock(return_value=mbs * 1024 * 1024))))))):
with patch.object(
UserState,
"max_monthly_mbs",
new_callable=PropertyMock,
return_value=max_mbs,
):
with patch.object(
us.db,
"query",
return_value=MagicMock(
filter=MagicMock(
return_value=MagicMock(
with_entities=MagicMock(
return_value=MagicMock(
scalar=MagicMock(
return_value=mbs * 1024 * 1024
)
)
)
)
)
),
):
assert us.has_quota_max_monthly_mbs("") == expected
@@ -400,10 +767,15 @@ def test_can_manually_trigger(user_state):
"group2": GroupInfo(manually_trigger_sheet=False),
}
with patch.object(UserState, 'permissions', new_callable=PropertyMock, return_value=permissions):
assert user_state.can_manually_trigger("group1") == True
assert user_state.can_manually_trigger("group2") == False
assert user_state.can_manually_trigger("group3") == False
with patch.object(
UserState,
"permissions",
new_callable=PropertyMock,
return_value=permissions,
):
assert user_state.can_manually_trigger("group1") is True
assert user_state.can_manually_trigger("group2") is False
assert user_state.can_manually_trigger("group3") is False
def test_is_sheet_frequency_allowed(user_state):
@@ -412,23 +784,44 @@ def test_is_sheet_frequency_allowed(user_state):
"group2": GroupInfo(sheet_frequency={"daily"}),
}
with patch.object(UserState, 'permissions', new_callable=PropertyMock, return_value=permissions):
assert user_state.is_sheet_frequency_allowed("group1", "daily") == True
assert user_state.is_sheet_frequency_allowed("group1", "hourly") == True
assert user_state.is_sheet_frequency_allowed("group1", "weekly") == False
assert user_state.is_sheet_frequency_allowed("group2", "hourly") == False
assert user_state.is_sheet_frequency_allowed("group2", "daily") == True
assert user_state.is_sheet_frequency_allowed("group3", "daily") == False
with patch.object(
UserState,
"permissions",
new_callable=PropertyMock,
return_value=permissions,
):
assert user_state.is_sheet_frequency_allowed("group1", "daily") is True
assert user_state.is_sheet_frequency_allowed("group1", "hourly") is True
assert (
user_state.is_sheet_frequency_allowed("group1", "weekly") is False
)
assert (
user_state.is_sheet_frequency_allowed("group2", "hourly") is False
)
assert user_state.is_sheet_frequency_allowed("group2", "daily") is True
assert user_state.is_sheet_frequency_allowed("group3", "daily") is False
def test_priority_group(user_state):
from app.web.utils.misc import convert_priority_to_queue_dict
with patch.object(UserState, 'user_groups', new_callable=PropertyMock, return_value=[
models.Group(id="group1", permissions={"priority": "high"}),
models.Group(id="group2", permissions={"priority": "medium"}),
models.Group(id="group3", permissions={"priority": "low"}),
]):
assert user_state.priority_group("group1") == convert_priority_to_queue_dict("high")
assert user_state.priority_group("group2") == convert_priority_to_queue_dict("medium")
assert user_state.priority_group("group3") == convert_priority_to_queue_dict("low")
assert user_state.priority_group("group4") == convert_priority_to_queue_dict("low")
with patch.object(
UserState,
"user_groups",
new_callable=PropertyMock,
return_value=[
models.Group(id="group1", permissions={"priority": "high"}),
models.Group(id="group2", permissions={"priority": "medium"}),
models.Group(id="group3", permissions={"priority": "low"}),
],
):
assert user_state.priority_group(
"group1"
) == convert_priority_to_queue_dict("high")
assert user_state.priority_group(
"group2"
) == convert_priority_to_queue_dict("medium")
assert user_state.priority_group(
"group3"
) == convert_priority_to_queue_dict("low")
assert user_state.priority_group(
"group4"
) == convert_priority_to_queue_dict("low")