Add tests to test the whole flow

Tests the device flow end to end

Author: duzumaki

tests/test_device.py

@@ -34,7 +34,7 @@ def setUpTestData(cls):
             name="test_client_credentials_app",
             user=cls.dev_user,
             client_type=Application.CLIENT_PUBLIC,
-            authorization_grant_type=Application.GRANT_CLIENT_CREDENTIALS,
+            authorization_grant_type=Application.GRANT_DEVICE_CODE,
             client_secret="abcdefghijklmnopqrstuvwxyz1234567890",
         )
 
@@ -101,6 +101,112 @@ def test_device_flow_authorization_initiation(self):
             "interval": 5,
         }
 
+    @mock.patch(
+        "oauthlib.oauth2.rfc8628.endpoints.device_authorization.generate_token",
+        lambda: "abc",
+    )
+    def test_device_flow_authorization_user_code_confirm_and_access_token(self):
+        """
+        1. User visits the /device endpoint in their browsers and submits the user code
+
+        the device and approve deny actions occur concurrently
+        (i.e the device is polling the token endpoint while the user
+        either approves or denies the device)
+
+        -2(3)-. User approves or denies the device
+        -3(2)-. Device polls the /token endpoint
+        """
+
+        # -----------------------
+        #  0: Setup device flow
+        # -----------------------
+        self.oauth2_settings.OAUTH_DEVICE_VERIFICATION_URI = "example.com/device"
+        self.oauth2_settings.OAUTH_DEVICE_USER_CODE_GENERATOR = lambda: "xyz"
+
+        request_data: dict[str, str] = {
+            "client_id": self.application.client_id,
+        }
+        request_as_x_www_form_urlencoded: str = urlencode(request_data)
+
+        django.http.response.JsonResponse = self.client.post(
+            reverse("oauth2_provider:device-authorization"),
+            data=request_as_x_www_form_urlencoded,
+            content_type="application/x-www-form-urlencoded",
+        )
+
+        # /device and /device_confirm require a user to be logged in
+        # to access it
+        UserModel.objects.create_user(
+            username="test_user_device_flow",
+            email="[email protected]",
+            password="password123",
+        )
+        self.client.login(username="test_user_device_flow", password="password123")
+
+        # --------------------------------------------------------------------------------
+        #  1. User visits the /device endpoint in their browsers and submits the user code
+        #  submits wrong code then right code
+        # --------------------------------------------------------------------------------
+
+        # 1. User visits the /device endpoint in their browsers and submits the user code
+        # (GET Request to load it)
+        get_response = self.client.get(reverse("oauth2_provider:device"))
+        assert get_response.status_code == 200
+        assert "form" in get_response.context  # Ensure the form is rendered in the context
+
+        # 1.1.0  User visits the /device endpoint in their browsers and submits wrong user code
+        with pytest.raises(oauth2_provider.models.Device.DoesNotExist):
+            self.client.post(
+                reverse("oauth2_provider:device"),
+                data={"user_code": "invalid_code"},
+            )
+
+        # 1.1.1: user submits valid user code
+        post_response_valid = self.client.post(
+            reverse("oauth2_provider:device"),
+            data={"user_code": "xyz"},
+        )
+
+        device_confirm_url = reverse("oauth2_provider:device-confirm", kwargs={"device_code": "abc"})
+        assert post_response_valid.status_code == 308  # Ensure it redirects with 308 status
+        assert post_response_valid["Location"] == device_confirm_url
+
+        # --------------------------------------------------------------------------------
+        # 2: We redirect to the accept/deny form (the user is still in their browser)
+        #  and approves
+        # --------------------------------------------------------------------------------
+        get_confirm = self.client.get(device_confirm_url)
+        assert get_confirm.status_code == 200
+
+        approve_response = self.client.post(device_confirm_url, data={"action": "accept"})
+        assert approve_response.status_code == 200
+        assert approve_response.content.decode() == "approved"
+
+        device = DeviceModel.objects.get(device_code="abc")
+        assert device.status == device.AUTHORIZED
+
+        # -------------------------
+        # 3: Device polls /token
+        # -------------------------
+        token_payload = {
+            "device_code": device.device_code,
+            "client_id": self.application.client_id,
+            "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+        }
+        token_response = self.client.post(
+            "/o/token/",
+            data=urlencode(token_payload),
+            content_type="application/x-www-form-urlencoded",
+        )
+
+        assert token_response.status_code == 200
+
+        token_data = token_response.json()
+
+        assert "access_token" in token_data
+        assert token_data["token_type"].lower() == "bearer"
+        assert "scope" in token_data
+
     @mock.patch(
         "oauthlib.oauth2.rfc8628.endpoints.device_authorization.generate_token",
         lambda: "abc",