aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMax Nanis2026-09-02 16:51:06 -0700
committerMax Nanis2026-09-02 16:51:06 -0700
commit97b14e2f133bda76f548ec1a522d9582c657d736 (patch)
tree8405bb2d1ed17a983f1a269ab25c20fc16ee15eb
parenta63de356383f3dc1d92c44e1167b27de05157146 (diff)
downloadgeneralresearch-97b14e2f133bda76f548ec1a522d9582c657d736.tar.gz
generalresearch-97b14e2f133bda76f548ec1a522d9582c657d736.zip
managers/gr/test_auth is all green ✅
-rw-r--r--generalresearch/thl_django/app/test_settings.py2
-rw-r--r--test_utils/managers/upk/conftest.py8
-rw-r--r--test_utils/models/gr/conftest.py254
-rw-r--r--tests/managers/gr/test_authentication.py115
4 files changed, 250 insertions, 129 deletions
diff --git a/generalresearch/thl_django/app/test_settings.py b/generalresearch/thl_django/app/test_settings.py
index c5df32a..2738aed 100644
--- a/generalresearch/thl_django/app/test_settings.py
+++ b/generalresearch/thl_django/app/test_settings.py
@@ -1,7 +1,7 @@
DATABASES = {
"default": {
"ENGINE": "django.db.backends.postgresql",
- "NAME": 'unittest-2026-09-02-a0310b',
+ "NAME": 'unittest-2026-09-02-77ae16',
"USER": 'jenkins',
"PASSWORD": '123456789',
"HOST": 'unittest-postgresql.fmt2.grl.internal',
diff --git a/test_utils/managers/upk/conftest.py b/test_utils/managers/upk/conftest.py
index f581278..23af1b3 100644
--- a/test_utils/managers/upk/conftest.py
+++ b/test_utils/managers/upk/conftest.py
@@ -13,11 +13,9 @@ from generalresearch.managers.thl.profiling.uqa import UQAManager
from generalresearch.managers.thl.profiling.user_upk import (
UserUpkManager,
)
-
-if TYPE_CHECKING:
- from generalresearch.models.thl.user import User
- from generalresearch.pg_helper import PostgresConfig
- from generalresearch.redis_helper import RedisConfig
+from generalresearch.models.thl.user import User
+from generalresearch.pg_helper import PostgresConfig
+from generalresearch.redis_helper import RedisConfig
@pytest.fixture(scope="session")
diff --git a/test_utils/models/gr/conftest.py b/test_utils/models/gr/conftest.py
index a73dd70..a5abf74 100644
--- a/test_utils/models/gr/conftest.py
+++ b/test_utils/models/gr/conftest.py
@@ -36,36 +36,6 @@ if TYPE_CHECKING:
# --- Factory / Database ---
-@pytest.fixture
-def gr_user_factory(gr_user_manager: GRUserManager) -> Callable[..., GRUser]:
-
- def _inner(
- sub: str | None = None,
- is_superuser: bool = False,
- ) -> GRUser:
- sub = sub or f"{uuid4().hex}-{uuid4().hex}"
-
- return gr_user_manager.create(
- sub=sub,
- is_superuser=is_superuser,
- )
-
- return _inner
-
-
-@pytest.fixture
-def gr_user_cache(
- gr_user: GRUser,
- gr_db: PostgresConfig,
- thl_web_rr: PostgresConfig,
- gr_redis_config: RedisConfig,
-) -> GRUser:
- gr_user.set_cache(
- pg_config=gr_db, thl_web_rr=thl_web_rr, redis_config=gr_redis_config
- )
- return gr_user
-
-
# --- Business Bank Account ---
@@ -98,7 +68,7 @@ def gr_business_bank_account_factory(
**kwargs,
)
else:
- raise ValueError("BusinessBankAccount Business not supported yet")
+ raise ValueError("Unsaved BusinessBankAccount not supported yet")
return _inner
@@ -115,16 +85,17 @@ def unsaved_gr_business_bank_account(
return gr_business_factory(save=False)
-# -----------------
+# --- Business Address ---
@pytest.fixture
def gr_business_address_factory(
- gr_bam: BusinessAddressManager,
+ gr_business_address_manager: BusinessAddressManager,
) -> Callable[..., BusinessAddress]:
def _inner(
business_id: PositiveInt,
+ save: bool = True,
uuid: UUIDStr | None = None,
line_1: str | None = None,
line_2: str | None = None,
@@ -133,7 +104,7 @@ def gr_business_address_factory(
postal_code: str | None = None,
phone_number: PhoneNumber | None = None,
country: str | None = None,
- ):
+ ) -> BusinessAddress:
uuid = uuid or uuid4().hex
line_1 = line_1 or "abc"
line_2 = line_2 or "bczx"
@@ -143,21 +114,48 @@ def gr_business_address_factory(
phone_number = None
country = country or "US"
- return gr_bam.create(
- business_id=business_id,
- uuid=uuid,
- line_1=line_1,
- line_2=line_2,
- city=city,
- state=state,
- postal_code=postal_code,
- phone_number=phone_number,
- country=country,
- )
+ if save:
+ return gr_business_address_manager.create(
+ business_id=business_id,
+ uuid=uuid,
+ line_1=line_1,
+ line_2=line_2,
+ city=city,
+ state=state,
+ postal_code=postal_code,
+ phone_number=phone_number,
+ country=country,
+ )
+ else:
+ raise ValueError("Unsaved BusinessAddress not supported yet")
return _inner
+# @pytest.fixture
+# def business_address(
+# gr_business: Business, business_address_manager: BusinessAddressManager
+# ) -> :
+# return business_address_manager.create_dummy(business_id=gr_business.id)
+
+
+@pytest.fixture
+def gr_business_address(
+ gr_business_address_factory: Callable[..., BusinessAddress],
+) -> BusinessAddress:
+ return gr_business_address_factory(save=True)
+
+
+@pytest.fixture
+def unsaved_gr_business_address(
+ gr_business_address_factory: Callable[..., BusinessAddress],
+) -> BusinessAddress:
+ return gr_business_address_factory(save=False)
+
+
+# --- Business ---
+
+
@pytest.fixture
def gr_business_factory(
gr_business_manager: BusinessManager,
@@ -194,37 +192,127 @@ def unsaved_gr_business(gr_business_factory: Callable[..., Business]) -> Busines
return gr_business_factory(save=False)
+# --- GR Team ---
+
+
@pytest.fixture
-def gr_team(
- gr_tm: TeamManager,
+def gr_team_factory(
+ gr_team_manager: TeamManager,
) -> Callable[..., Team]:
- def _inner(uuid: UUIDStr | None = None, name: str | None = None) -> Team:
- uuid = uuid or uuid4().hex
- name = name or f"name-{uuid4().hex[:12]}"
+ def _inner(
+ save: bool = True,
+ uuid: UUIDStr | None = None,
+ name: str | None = None,
+ **kwargs,
+ ) -> Team:
+
+ if save:
+ return gr_team_manager.create(uuid=uuid, name=name, **kwargs)
- return gr_tm.create(uuid=uuid, name=name)
+ else:
+ raise ValueError("BusinessBankAccount Business not supported yet")
return _inner
-@pytest.fixture()
-def gr_user_token(
- gr_user: GRUser, gr_tm: GRTokenManager, gr_db: PostgresConfig
-) -> GRToken:
- gr_tm.create(user_id=gr_user.id)
- gr_user.prefetch_token(pg_config=gr_db)
+@pytest.fixture
+def gr_team(gr_team_factory: Callable[..., Team]) -> Team:
+ return gr_team_factory(save=True)
+
+
+@pytest.fixture
+def unsaved_gr_team(
+ gr_team_factory: Callable[..., Team],
+) -> Team:
+ return gr_team_factory(save=False)
- res = gr_user.token
- assert res is not None, "GRToken should exist after creation and prefetching"
- return res
+
+# --- GR User ---
@pytest.fixture
-def business_address(
- gr_business: Business, business_address_manager: BusinessAddressManager
-) -> BusinessAddress:
- return business_address_manager.create_dummy(business_id=gr_business.id)
+def gr_user_factory(gr_user_manager: GRUserManager) -> Callable[..., GRUser]:
+
+ def _inner(
+ save: bool = True,
+ sub: str | None = None,
+ is_superuser: bool = False,
+ ) -> GRUser:
+ sub = sub or f"{uuid4().hex}-{uuid4().hex}"
+
+ if save:
+ return gr_user_manager.create(
+ sub=sub,
+ is_superuser=is_superuser,
+ )
+ else:
+ raise ValueError("Unsaved GR User not supported yet")
+
+ return _inner
+
+
+@pytest.fixture
+def gr_user_cache(
+ gr_user: GRUser,
+ gr_db: PostgresConfig,
+ thl_web_rr: PostgresConfig,
+ gr_redis_config: RedisConfig,
+) -> GRUser:
+ gr_user.set_cache(
+ pg_config=gr_db, thl_web_rr=thl_web_rr, redis_config=gr_redis_config
+ )
+ return gr_user
+
+
+@pytest.fixture
+def gr_user(gr_user_factory: Callable[..., GRUser]) -> GRUser:
+ return gr_user_factory(save=True)
+
+
+@pytest.fixture
+def unsaved_gr_user(
+ gr_user_factory: Callable[..., GRUser],
+) -> GRUser:
+ return gr_user_factory(save=False)
+
+
+# --- GR User Token ---
+
+
+@pytest.fixture
+def gr_user_token_factory(
+ gr_user: GRUser, gr_user_token_manager: GRUser, gr_db: PostgresConfig
+) -> Callable[..., GRToken]:
+
+ def _inner(
+ save: bool = True,
+ ) -> GRToken:
+
+ if save:
+ gr_user_token_manager.create(user_id=gr_user.id)
+ gr_user.prefetch_token(pg_config=gr_db)
+
+ res = gr_user.token
+ assert (
+ res is not None
+ ), "GRToken should exist after creation and prefetching"
+ return res
+
+ else:
+ raise ValueError("Unsaved GR User not supported yet")
+
+ return _inner
+
+
+@pytest.fixture
+def gr_user_token(gr_user_token_factory: Callable[..., GRToken]) -> GRToken:
+ return gr_user_token_factory(save=True)
+
+
+@pytest.fixture
+def unsaved_gr_user_token(gr_user_token_factory: Callable[..., GRToken]) -> GRToken:
+ return gr_user_token_factory(save=False)
@pytest.fixture()
@@ -232,26 +320,32 @@ def gr_user_token_header(gr_user_token: GRToken) -> dict[str, str]:
return gr_user_token.auth_header
-@pytest.fixture(scope="function")
-def membership(team: Team, gr_user: GRUser, team_manager: TeamManager) -> Membership:
- assert team.id, "Team must be saved"
- assert gr_user.id, "GRUser must be saved"
- return team_manager.add_user(team=team, gr_user=gr_user)
+# --- GR Membership ---
-@pytest.fixture(scope="function")
-def membership_factory(
- team: Team,
+@pytest.fixture()
+def gr_membership_factory(
+ gr_team: Team,
gr_user: GRUser,
- membership_manager: MembershipManager,
- team_manager: TeamManager,
- gr_um: GRUserManager,
+ gr_membership_manager: MembershipManager,
) -> Callable[..., Membership]:
- def _inner(**kwargs) -> Membership:
- _team = kwargs.get("team", team_manager.create_dummy())
- _gr_user = kwargs.get("gr_user", gr_um.create_dummy())
-
- return membership_manager.create(team=_team, gr_user=_gr_user)
+ def _inner(save: bool = True, **kwargs) -> Membership:
+ if save:
+ return gr_membership_manager.create(team=gr_team, gr_user=gr_user, **kwargs)
+ else:
+ raise ValueError("Unsaved GR Membership not supported yet")
return _inner
+
+
+@pytest.fixture()
+def gr_membership(gr_membership_factory: Callable[..., Membership]) -> Membership:
+ return gr_membership_factory(save=True)
+
+
+@pytest.fixture()
+def unsaved_gr_membership(
+ gr_membership_factory: Callable[..., Membership],
+) -> Membership:
+ return gr_membership_factory(save=False)
diff --git a/tests/managers/gr/test_authentication.py b/tests/managers/gr/test_authentication.py
index b9f43a6..0bcabc5 100644
--- a/tests/managers/gr/test_authentication.py
+++ b/tests/managers/gr/test_authentication.py
@@ -1,117 +1,146 @@
import logging
+from collections.abc import Callable
from uuid import uuid4
import pytest
-from generalresearch.models.gr.authentication import GRUser
+from generalresearch.managers.gr.authentication import GRTokenManager, GRUserManager
+from generalresearch.managers.gr.team import TeamManager
+from generalresearch.models.gr.authentication import GRToken, GRUser
+from generalresearch.pg_helper import PostgresConfig
+from generalresearch.redis_helper import RedisConfig
SSO_ISSUER = ""
class TestGRUserManager:
- def test_create(self, gr_um):
-
- user: GRUser = gr_um.create_dummy()
- instance = gr_um.get_by_id(user.id)
- assert user.id == instance.id
+ def test_create(self, gr_user: GRUser, gr_user_manager: GRUserManager):
+ instance = gr_user_manager.get_by_id(gr_user.id)
+ assert isinstance(instance, GRUser)
+ assert gr_user.id == instance.id
- instance2 = gr_um.get_by_id(user.id)
- assert user.model_dump_json() == instance2.model_dump_json()
+ instance2 = gr_user_manager.get_by_id(gr_user.id)
+ assert isinstance(instance2, GRUser)
+ assert gr_user.model_dump_json() == instance2.model_dump_json()
- def test_get_by_id(self, gr_user, gr_um):
+ def test_get_by_id(self, gr_user: GRUser, gr_user_manager: GRUserManager):
with pytest.raises(expected_exception=ValueError) as cm:
- gr_um.get_by_id(gr_user_id=999_999_999)
+ gr_user_manager.get_by_id(gr_user_id=999_999_999)
assert "GRUser not found" in str(cm.value)
- instance = gr_um.get_by_id(gr_user_id=gr_user.id)
+ instance = gr_user_manager.get_by_id(gr_user_id=gr_user.id)
+ assert isinstance(instance, GRUser)
assert instance.sub == gr_user.sub
- def test_get_by_sub(self, gr_user, gr_um):
+ def test_get_by_sub(self, gr_user: GRUser, gr_user_manager: GRUserManager):
with pytest.raises(expected_exception=ValueError) as cm:
- gr_um.get_by_sub(sub=uuid4().hex)
+ gr_user_manager.get_by_sub(sub=uuid4().hex)
assert "GRUser not found" in str(cm.value)
- instance = gr_um.get_by_sub(sub=gr_user.sub)
+ instance = gr_user_manager.get_by_sub(sub=gr_user.sub)
+ assert isinstance(instance, GRUser)
assert instance.id == gr_user.id
- def test_get_by_sub_or_create(self, gr_user, gr_um):
+ def test_get_by_sub_or_create(
+ self, gr_user: GRUser, gr_user_manager: GRUserManager
+ ):
sub = f"{uuid4().hex}-{uuid4().hex}"
with pytest.raises(expected_exception=ValueError) as cm:
- gr_um.get_by_sub(sub=sub)
+ gr_user_manager.get_by_sub(sub=sub)
assert "GRUser not found" in str(cm.value)
- instance = gr_um.get_by_sub_or_create(sub=sub)
+ instance = gr_user_manager.get_by_sub_or_create(sub=sub)
assert isinstance(instance, GRUser)
assert instance.sub == sub
- def test_get_all(self, gr_um):
- res1 = gr_um.get_all()
+ def test_get_all(
+ self, gr_user_factory: Callable[..., GRUser], gr_user_manager: GRUserManager
+ ):
+ res1 = gr_user_manager.get_all()
assert isinstance(res1, list)
- gr_um.create_dummy()
- res2 = gr_um.get_all()
+ gr_user_factory(save=True)
+ res2 = gr_user_manager.get_all()
assert len(res1) == len(res2) - 1
- def test_get_by_team(self, gr_um):
- res = gr_um.get_by_team(team_id=999_999_999)
+ def test_get_by_team(self, gr_user_manager: GRUserManager):
+ res = gr_user_manager.get_by_team(team_id=999_999_999)
assert isinstance(res, list)
assert res == []
- def test_list_product_uuids(self, caplog, gr_user, gr_um, thl_web_rr):
+ def test_list_product_uuids(
+ self,
+ caplog,
+ gr_user: GRUser,
+ gr_user_manager: GRUserManager,
+ thl_web_rr: PostgresConfig,
+ ):
with caplog.at_level(logging.WARNING):
- gr_um.list_product_uuids(user=gr_user, thl_pg_config=thl_web_rr)
+ gr_user_manager.list_product_uuids(user=gr_user, thl_pg_config=thl_web_rr)
assert "prefetch not run" in caplog.text
class TestGRTokenManager:
- def test_create(self, gr_user, gr_tm):
- assert gr_tm.create(user_id=gr_user.id) is None
+ def test_create(self, gr_user: GRUser, gr_team_manager: TeamManager):
+ assert gr_team_manager.create(user_id=gr_user.id) is None
- token = gr_tm.get_by_user_id(user_id=gr_user.id)
+ token = gr_team_manager.get_by_user_id(user_id=gr_user.id)
assert gr_user.id == token.user_id
- def test_get_by_user_id(self, gr_user, gr_tm):
- assert gr_tm.create(user_id=gr_user.id) is None
+ def test_get_by_user_id(self, gr_user: GRUser, gr_team_manager: TeamManager):
+ assert gr_team_manager.create(user_id=gr_user.id) is None
- token = gr_tm.get_by_user_id(user_id=gr_user.id)
+ token = gr_team_manager.get_by_user_id(user_id=gr_user.id)
assert gr_user.id == token.user_id
- def test_prefetch_user(self, gr_user, gr_tm, gr_db, gr_redis_config):
- from generalresearch.models.gr.authentication import GRToken
+ def test_prefetch_user(
+ self,
+ gr_user: GRUser,
+ gr_team_manager: TeamManager,
+ gr_db: PostgresConfig,
+ gr_redis_config: RedisConfig,
+ ):
- gr_tm.create(user_id=gr_user.id)
+ gr_team_manager.create(user_id=gr_user.id)
- token: GRToken = gr_tm.get_by_user_id(user_id=gr_user.id)
+ token: GRToken = gr_team_manager.get_by_user_id(user_id=gr_user.id)
assert token.user is None
token.prefetch_user(pg_config=gr_db, redis_config=gr_redis_config)
assert token.user.id == gr_user.id
- def test_get_by_key(self, gr_user, gr_um, gr_tm):
- gr_tm.create(user_id=gr_user.id)
- token = gr_tm.get_by_user_id(user_id=gr_user.id)
+ def test_get_by_key(
+ self,
+ gr_user: GRUser,
+ gr_team_manager: TeamManager,
+ ):
+ gr_team_manager.create(user_id=gr_user.id)
+ token = gr_team_manager.get_by_user_id(user_id=gr_user.id)
- instance = gr_tm.get_by_key(api_key=token.key)
+ instance = gr_team_manager.get_by_key(api_key=token.key)
assert token.created == instance.created
# Search for non-existent key
with pytest.raises(expected_exception=Exception) as cm:
- gr_tm.get_by_key(api_key=uuid4().hex)
+ gr_team_manager.get_by_key(api_key=uuid4().hex)
assert "No GRUser with token of " in str(cm.value)
@pytest.mark.skip(reason="no idea how to actually test this...")
- def test_get_by_sso_key(self, gr_user, gr_um, gr_tm, gr_redis_config):
- from generalresearch.models.gr.authentication import GRToken
+ def test_get_by_sso_key(
+ self,
+ gr_team_manager: TeamManager,
+ gr_redis_config: RedisConfig,
+ ):
api_key = "..."
jwks = {
# ...
}
- instance = gr_tm.get_by_key(
+ instance = gr_team_manager.get_by_key(
api_key=api_key,
jwks=jwks,
audience="...",