Update linkedin

src/django_project/core/settings.py

@@ -289,6 +289,9 @@
 GEMINI_API_KEY: str | None = env.str("GEMINI_API_KEY", None)
 GOOGLE_ANALYTICS_TAG_ID: str | None = env.str("GOOGLE_ANALYTICS_TAG_ID", None)
 LINKEDIN_ACCESS_TOKEN: str | None = env.str("LINKEDIN_ACCESS_TOKEN", None)
+LINKEDIN_CLIENT_ID: str | None = env.str("LINKEDIN_CLIENT_ID", None)
+LINKEDIN_CLIENT_SECRET: str | None = env.str("LINKEDIN_CLIENT_SECRET", None)
 LINKEDIN_ORGANIZATION_URN: str | None = env.str("LINKEDIN_ORGANIZATION_URN", None)
+LINKEDIN_REFRESH_TOKEN: str | None = env.str("LINKEDIN_REFRESH_TOKEN", None)
 SPUG_API_TOKEN: str | None = env.str("SPUG_API_TOKEN", None)
 SPUG_API_URL: str | None = env.str("SPUG_API_URL", None)

src/django_project/core/urls.py

@@ -18,10 +18,12 @@
 from django.conf import settings
 from django.contrib import admin
 from django.urls import include, path
+from web.views import linkedin_oauth_callback
 
 urlpatterns: list = [
     # Django provided URLs
     path("console/", admin.site.urls),
+    path("accounts/oidc/linkedin/login/callback/", linkedin_oauth_callback, name="linkedin_oauth_callback"),
     path("accounts/", include("django.contrib.auth.urls")),
     # 3rd party URLs
     path("handyhelpers/", include("handyhelpers.urls")),

src/django_project/tests/unit/core/views/test_views.py

@@ -35,3 +35,24 @@ def test_post(self) -> None:
         """verify call to HostView view"""
         response: HttpResponse = self.client.post(reverse("host"))
         self.assertEqual(response.status_code, 405)
+
+
+class LinkedInOAuthCallbackTests(TestCase):
+    def test_get_with_code(self) -> None:
+        response: HttpResponse = self.client.get(
+            reverse("linkedin_oauth_callback"),
+            {"code": "example-code", "state": "example-state"},
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response["content-type"], "text/plain")
+        self.assertIn("code: example-code", response.content.decode("utf-8"))
+        self.assertIn("state: example-state", response.content.decode("utf-8"))
+
+    def test_get_with_error(self) -> None:
+        response: HttpResponse = self.client.get(
+            reverse("linkedin_oauth_callback"),
+            {"error": "access_denied", "error_description": "user denied"},
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(response["content-type"], "text/plain")
+        self.assertIn("error: access_denied", response.content.decode("utf-8"))

src/django_project/tests/unit/web/test_linkedin_notifier.py

@@ -0,0 +1,163 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import requests
+
+BASE_DIR = Path(__file__).parents[4]
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")
+os.environ.setdefault("ENV_PATH", f"{BASE_DIR}/envs/.env.test")
+
+from web.utilities.notifiers.linkedin import LinkedInOrganizationClient
+
+
+class DummyCredentialManager:
+    def __init__(self, credential):
+        self.credential = credential
+
+    def select_for_update(self):
+        return self
+
+    def get(self, pk):
+        return self.credential
+
+
+class DummyCredential:
+    objects = None
+
+    def __init__(self):
+        self.pk = 1
+        self.access_token = "old-token"
+        self.refresh_token = "refresh-token"
+        self.access_token_expires_at = None
+        self.refresh_token_expires_at = None
+        self.save = Mock()
+
+
+class LinkedInOrganizationClientTests(unittest.TestCase):
+    def build_client(self, env_path: str | None = None, access_token: str | None = "old-token") -> LinkedInOrganizationClient:
+        return LinkedInOrganizationClient(
+            access_token=access_token,
+            organization_urn="urn:li:organization:107506588",
+            client_id="client-id",
+            client_secret="client-secret",
+            refresh_token="refresh-token",
+            env_path=env_path,
+        )
+
+    def test_refresh_access_token_updates_settings_and_env_file(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_env = Path(temp_dir) / ".env.test"
+            temp_env.write_text(
+                'LINKEDIN_ACCESS_TOKEN="old-token"\nLINKEDIN_REFRESH_TOKEN="refresh-token"\n',
+                encoding="utf-8",
+            )
+            client = self.build_client(env_path=str(temp_env))
+
+            response = Mock()
+            response.json.return_value = {
+                "access_token": "new-token",
+                "refresh_token": "new-refresh-token",
+            }
+            response.raise_for_status.return_value = None
+
+            with (
+                patch("web.utilities.notifiers.linkedin.requests.post", return_value=response) as mock_post,
+                patch("web.utilities.notifiers.linkedin.settings") as mock_settings,
+            ):
+                client.refresh_access_token()
+
+            self.assertEqual(client.access_token, "new-token")
+            self.assertEqual(client.refresh_token, "new-refresh-token")
+            self.assertIn('LINKEDIN_ACCESS_TOKEN="new-token"', temp_env.read_text(encoding="utf-8"))
+            self.assertIn('LINKEDIN_REFRESH_TOKEN="new-refresh-token"', temp_env.read_text(encoding="utf-8"))
+            mock_post.assert_called_once()
+            self.assertEqual(mock_settings.LINKEDIN_ACCESS_TOKEN, "new-token")
+            self.assertEqual(mock_settings.LINKEDIN_REFRESH_TOKEN, "new-refresh-token")
+
+    def test_post_retries_once_after_auth_failure(self):
+        client = self.build_client()
+
+        auth_failure_response = Mock(status_code=401)
+        auth_failure_response.raise_for_status.side_effect = requests.HTTPError(response=auth_failure_response)  # type: ignore[name-defined]
+
+        refresh_response = Mock()
+        refresh_response.json.return_value = {
+            "access_token": "new-token",
+            "refresh_token": "new-refresh-token",
+        }
+        refresh_response.raise_for_status.return_value = None
+
+        success_response = Mock(status_code=201)
+        success_response.raise_for_status.return_value = None
+
+        with (
+            patch(
+                "web.utilities.notifiers.linkedin.requests.post",
+                side_effect=[auth_failure_response, refresh_response, success_response],
+            ) as mock_post,
+            patch("web.utilities.notifiers.linkedin.settings"),
+        ):
+            response = client.post_organization_post("hello world")
+
+        self.assertIs(response, success_response)
+        self.assertEqual(client.access_token, "new-token")
+        self.assertEqual(mock_post.call_count, 3)
+
+    def test_refresh_access_token_updates_db_credential_when_present(self):
+        credential = DummyCredential()
+        credential.__class__.objects = DummyCredentialManager(credential)
+
+        client = LinkedInOrganizationClient(
+            access_token="old-token",
+            organization_urn="urn:li:organization:107506588",
+            client_id="client-id",
+            client_secret="client-secret",
+            refresh_token="refresh-token",
+            credential=credential,
+        )
+
+        refresh_response = Mock()
+        refresh_response.json.return_value = {
+            "access_token": "new-token",
+            "refresh_token": "new-refresh-token",
+            "expires_in": 3600,
+            "refresh_token_expires_in": 7200,
+        }
+        refresh_response.raise_for_status.return_value = None
+
+        with (
+            patch("web.utilities.notifiers.linkedin.requests.post", return_value=refresh_response),
+            patch("web.utilities.notifiers.linkedin.settings"),
+        ):
+            client.refresh_access_token()
+
+        self.assertEqual(credential.access_token, "new-token")
+        self.assertEqual(credential.refresh_token, "new-refresh-token")
+        credential.save.assert_called_once()
+
+    def test_missing_tokens_can_be_bootstrapped_from_refresh_credentials(self):
+        client = self.build_client(access_token=None)
+
+        refresh_response = Mock()
+        refresh_response.json.return_value = {
+            "access_token": "new-token",
+            "refresh_token": "new-refresh-token",
+        }
+        refresh_response.raise_for_status.return_value = None
+
+        success_response = Mock(status_code=201)
+        success_response.raise_for_status.return_value = None
+
+        with (
+            patch(
+                "web.utilities.notifiers.linkedin.requests.post",
+                side_effect=[refresh_response, success_response],
+            ),
+            patch("web.utilities.notifiers.linkedin.settings"),
+        ):
+            client.post_organization_post("hello world")
+
+        self.assertEqual(client.access_token, "new-token")

src/django_project/tests/unit/web/test_linkedin_oauth_command.py

@@ -0,0 +1,79 @@
+import os
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+BASE_DIR = Path(__file__).parents[4]
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")
+os.environ.setdefault("ENV_PATH", f"{BASE_DIR}/envs/.env.test")
+
+from web.management.commands.linkedin_oauth import Command
+
+
+class LinkedInOAuthCommandTests(unittest.TestCase):
+    def test_show_url_outputs_authorization_url(self):
+        command = Command()
+        command.stdout = Mock()
+
+        with (
+            patch("web.management.commands.linkedin_oauth.apps.get_model") as mock_get_model,
+            patch("web.management.commands.linkedin_oauth.settings") as mock_settings,
+            patch("web.management.commands.linkedin_oauth.LinkedInOrganizationClient") as mock_client_class,
+        ):
+            mock_settings.LINKEDIN_CLIENT_ID = "client-id"
+            mock_settings.LINKEDIN_CLIENT_SECRET = "client-secret"
+            mock_settings.LINKEDIN_ACCESS_TOKEN = None
+            mock_settings.LINKEDIN_ORGANIZATION_URN = "urn:li:organization:107506588"
+            mock_settings.LINKEDIN_REFRESH_TOKEN = None
+            mock_settings.ENV_PATH = "/tmp/.env.test"
+
+            mock_credential = Mock(access_token=None, refresh_token=None)
+            mock_get_model.return_value.objects.get_or_create.return_value = (mock_credential, True)
+
+            mock_client = mock_client_class.return_value
+            mock_client.build_authorization_url.return_value = "https://www.linkedin.com/oauth/v2/authorization?x=1"
+
+            command.handle(redirect_uri="https://example.com/callback", scope=command.default_scope, state=None, code=None)
+
+        command.stdout.write.assert_any_call("Open this URL in a browser and complete the LinkedIn consent flow:")
+        command.stdout.write.assert_any_call("https://www.linkedin.com/oauth/v2/authorization?x=1")
+
+    def test_code_exchange_stores_tokens(self):
+        command = Command()
+        command.stdout = Mock()
+
+        with (
+            patch("web.management.commands.linkedin_oauth.apps.get_model") as mock_get_model,
+            patch("web.management.commands.linkedin_oauth.settings") as mock_settings,
+            patch("web.management.commands.linkedin_oauth.LinkedInOrganizationClient") as mock_client_class,
+        ):
+            mock_settings.LINKEDIN_CLIENT_ID = "client-id"
+            mock_settings.LINKEDIN_CLIENT_SECRET = "client-secret"
+            mock_settings.LINKEDIN_ACCESS_TOKEN = None
+            mock_settings.LINKEDIN_ORGANIZATION_URN = "urn:li:organization:107506588"
+            mock_settings.LINKEDIN_REFRESH_TOKEN = None
+            mock_settings.ENV_PATH = "/tmp/.env.test"
+
+            mock_credential = Mock(access_token=None, refresh_token=None)
+            mock_get_model.return_value.objects.get_or_create.return_value = (mock_credential, True)
+
+            mock_client = mock_client_class.return_value
+            mock_client.exchange_authorization_code.return_value = {
+                "access_token": "new-token",
+                "refresh_token": "new-refresh-token",
+                "expires_in": 5184000,
+                "refresh_token_expires_in": 31536000,
+            }
+
+            command.handle(
+                redirect_uri="https://example.com/callback",
+                scope=command.default_scope,
+                state=None,
+                code="auth-code",
+                show_url=False,
+            )
+
+        mock_client.exchange_authorization_code.assert_called_once_with(
+            code="auth-code",
+            redirect_uri="https://example.com/callback",
+        )

src/django_project/web/admin.py

@@ -1,7 +1,14 @@
 from django.contrib import admin
 
 # import models
-from web.models import Event, Link, SocialPlatform, Tag, TechGroup
+from web.models import (
+    Event,
+    IntegrationCredential,
+    Link,
+    SocialPlatform,
+    Tag,
+    TechGroup,
+)
 
 
 class TagAdmin(admin.ModelAdmin):
@@ -20,6 +27,18 @@ class SocialPlatformAdmin(admin.ModelAdmin):
     list_filter = ["enabled"]
 
 
+class IntegrationCredentialAdmin(admin.ModelAdmin):
+    list_display = [
+        "id",
+        "provider",
+        "access_token_expires_at",
+        "refresh_token_expires_at",
+        "created_at",
+        "updated_at",
+    ]
+    search_fields = ["id", "provider"]
+
+
 class TechGroupAdmin(admin.ModelAdmin):
     list_display = ["id", "name", "description", "enabled", "platform", "icon", "image", "created_at", "updated_at"]
     search_fields = ["id", "name", "description", "icon", "image"]
@@ -61,5 +80,6 @@ class EventAdmin(admin.ModelAdmin):
 admin.site.register(Tag, TagAdmin)
 admin.site.register(Link, LinkAdmin)
 admin.site.register(SocialPlatform, SocialPlatformAdmin)
+admin.site.register(IntegrationCredential, IntegrationCredentialAdmin)
 admin.site.register(TechGroup, TechGroupAdmin)
 admin.site.register(Event, EventAdmin)

src/django_project/web/management/__init__.py

@@ -0,0 +1 @@
+

src/django_project/web/management/commands/__init__.py

@@ -0,0 +1 @@
+