@test@example.com, welcome!
" -async def stator_process_tasks(stator): - """ - Guarded wrapper to simply async_to_sync and ensure all stator tasks are - run to completion without blocking indefinitely. - """ - await asyncio.wait_for(stator.fetch_and_process_tasks(), timeout=1) - for _ in range(100): - if not stator.tasks: - break - stator.remove_completed_tasks() - await asyncio.sleep(0.01) - - @pytest.mark.django_db -def test_post_transitions(identity, stator_runner): +def test_post_transitions(identity, stator): # Create post post = Post.objects.create( @@ -153,18 +137,18 @@ def test_post_transitions(identity, stator_runner): ) # Test: | --> new --> fanned_out assert post.state == str(PostStates.new) - async_to_sync(stator_process_tasks)(stator_runner) + stator.run_single_cycle_sync() post = Post.objects.get(id=post.id) assert post.state == str(PostStates.fanned_out) # Test: fanned_out --> (forced) edited --> edited_fanned_out Post.transition_perform(post, PostStates.edited) - async_to_sync(stator_process_tasks)(stator_runner) + stator.run_single_cycle_sync() post = Post.objects.get(id=post.id) assert post.state == str(PostStates.edited_fanned_out) # Test: edited_fanned_out --> (forced) deleted --> deleted_fanned_out Post.transition_perform(post, PostStates.deleted) - async_to_sync(stator_process_tasks)(stator_runner) + stator.run_single_cycle_sync() post = Post.objects.get(id=post.id) assert post.state == str(PostStates.deleted_fanned_out) diff --git a/tests/conftest.py b/tests/conftest.py index a3feaca..80622f0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -60,7 +60,10 @@ def config_system(keypair): system_actor_private_key=keypair["private_key"], system_actor_public_key=keypair["public_key"], ) + Config.__forced__ = True yield Config.system + Config.__forced__ = False + del Config.system @pytest.fixture @@ -126,7 +129,7 @@ def remote_identity() -> Identity: @pytest.fixture -def stator_runner(config_system) -> StatorRunner: +def stator(config_system) -> StatorRunner: """ Return an initialized StatorRunner for tests that need state transitioning to happen. diff --git a/tests/users/views/test_auth.py b/tests/users/views/test_auth.py index f3a34c0..6dd1010 100644 --- a/tests/users/views/test_auth.py +++ b/tests/users/views/test_auth.py @@ -1,60 +1,119 @@ -from unittest import mock - import pytest +from django.core import mail +from pytest_django.asserts import assertContains, assertNotContains -from core.models import Config -from users.models import User - - -@pytest.fixture -def config_system(): - # TODO: Good enough for now, but a better Config mocking system is needed - result = Config.load_system() - with mock.patch("core.models.Config.load_system", return_value=result): - yield result +from users.models import Invite, User @pytest.mark.django_db def test_signup_disabled(client, config_system): + """ + Tests that disabling signup takes effect + """ # Signup disabled and no signup text config_system.signup_allowed = False - resp = client.get("/auth/signup/") - assert resp.status_code == 200 - content = str(resp.content) - assert "Not accepting new users at this time" in content - assert "" not in content + response = client.get("/auth/signup/") + assertContains(response, "Not accepting new users at this time", status_code=200) + assertNotContains(response, "") # Signup disabled with signup text configured config_system.signup_text = "Go away!!!!!!" - resp = client.get("/auth/signup/") - assert resp.status_code == 200 - content = str(resp.content) - assert "Go away!!!!!!" in content + response = client.get("/auth/signup/") + assertContains(response, "Go away!!!!!!", status_code=200) # Ensure direct POST doesn't side step guard - resp = client.post( + response = client.post( "/auth/signup/", data={"email": "test_signup_disabled@example.org"} ) - assert resp.status_code == 200 + assert response.status_code == 200 assert not User.objects.filter(email="test_signup_disabled@example.org").exists() # Signup enabled config_system.signup_allowed = True - resp = client.get("/auth/signup/") - assert resp.status_code == 200 - content = str(resp.content) - assert "Not accepting new users at this time" not in content - assert "" in content + response = client.get("/auth/signup/") + assertContains(response, "", status_code=200) + assertNotContains(response, "Not accepting new users at this time") @pytest.mark.django_db def test_signup_invite_only(client, config_system): + """ + Tests that invite codes work with signup + """ config_system.signup_allowed = True config_system.signup_invite_only = True - resp = client.get("/auth/signup/") - assert resp.status_code == 200 - content = str(resp.content) - assert 'name="invite_code"' in content + # Try to sign up without an invite code + response = client.post("/auth/signup/", {"email": "random@example.com"}) + assertNotContains(response, "Email Sent", status_code=200) + + # Make an invite code for any email + invite_any = Invite.create_random() + response = client.post( + "/auth/signup/", + {"email": "random@example.com", "invite_code": invite_any.token}, + ) + assertNotContains(response, "not a valid invite") + assertContains(response, "Email Sent", status_code=200) + + # Make sure you can't reuse an invite code + response = client.post( + "/auth/signup/", + {"email": "random2@example.com", "invite_code": invite_any.token}, + ) + assertNotContains(response, "Email Sent", status_code=200) + + # Make an invite code for a specific email + invite_specific = Invite.create_random(email="special@example.com") + response = client.post( + "/auth/signup/", + {"email": "random3@example.com", "invite_code": invite_specific.token}, + ) + assertContains(response, "valid invite code for this email", status_code=200) + assertNotContains(response, "Email Sent") + response = client.post( + "/auth/signup/", + {"email": "special@example.com", "invite_code": invite_specific.token}, + ) + assertContains(response, "Email Sent", status_code=200) + + +@pytest.mark.django_db +def test_signup_policy(client, config_system): + """ + Tests that you must agree to policies to sign up + """ + config_system.signup_allowed = True + config_system.signup_invite_only = False + + # Make sure we can sign up when there are no policies + response = client.post("/auth/signup/", {"email": "random@example.com"}) + assertContains(response, "Email Sent", status_code=200) + + # Make sure that's then denied when we have a policy in place + config_system.policy_rules = "You must love unit tests" + response = client.post("/auth/signup/", {"email": "random2@example.com"}) + assertContains(response, "field is required", status_code=200) + assertNotContains(response, "Email Sent") + + +@pytest.mark.django_db +def test_signup_email(client, config_system, stator): + """ + Tests that you can sign up and get an email sent to you + """ + config_system.signup_allowed = True + config_system.signup_invite_only = False + + # Sign up with a user + response = client.post("/auth/signup/", {"email": "random@example.com"}) + assertContains(response, "Email Sent", status_code=200) + + # Verify that made a user object and a password reset + user = User.objects.get(email="random@example.com") + assert user.password_resets.exists() - # TODO: Actually test this + # Run Stator and verify it sends the email + assert len(mail.outbox) == 0 + stator.run_single_cycle_sync() + assert len(mail.outbox) == 1 -- cgit v1.2.3