modify changes for webhooks and unit testing

Author: kanmitcode

integrations/github-cloud/.env.example

@@ -5,4 +5,5 @@ OCEAN__PORT__BASE_URL=https://api.getport.io
 OCEAN__EVENT_LISTENER__TYPE=POLLING
 OCEAN__INITIALIZE_PORT_RESOURCES=true
 OCEAN__INTEGRATION__CONFIG__GITHUB_BASE_URL=https://api.github.com
-OCEAN__INTEGRATION__CONFIG__GITHUB_ACCESS_TOKEN=github_access_token
+OCEAN__INTEGRATION__CONFIG__GITHUB_ACCESS_TOKEN=github-access-token
+OCEAN__INTEGRATION__CONFIG__APP_HOST=app-host

integrations/github-cloud/.port/spec.yaml

@@ -15,6 +15,10 @@ configurations:
     sensitive: true
     description: "Token for authenticating with GitHub Cloud"
     required: false
+  - name: appHost
+    type: string
+    description: "App Host"
+    required: true
   - name: githubBaseUrl
     type: url
     description: "Base URL for GitHub API"

integrations/github-cloud/client.py

@@ -16,6 +16,7 @@ class GithubHandler:
     def __init__(self) -> None:
         self.token = ocean.integration_config["github_access_token"]
         self.base_url = ocean.integration_config["github_base_url"]
+        self.app_host = ocean.integration_config['app_host']
         self.client = http_async_client
         self.headers = {
             'Authorization': f'token {self.token}',
@@ -39,7 +40,11 @@ async def fetch_with_retry(self, url: str, retries: int = 3, backoff_factor: int
             response = await self.client.get(url, headers=self.headers)
             match response.status_code:
                 case 200:
-                    return response.json()
+                    # Check if response.json is awaitable
+                    if asyncio.iscoroutinefunction(response.json):
+                        return await response.json()
+                    else:
+                        return response.json()
                 case 429:  # Rate limit exceeded
                     retry_after = int(response.headers.get("Retry-After", backoff_factor))
                     logger.warning(f"Rate limit exceeded. Retrying after {retry_after} seconds.")

integrations/github-cloud/github_cloud/__init__.py

@@ -0,0 +1,82 @@
+import logging
+import sys
+import os
+
+sys.path.append(os.path.abspath(os.path.dirname(__file__)))
+
+from port_ocean.context.ocean import ocean
+from client import GithubHandler
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+async def manage_webhooks() -> None:
+    """Manage webhooks for all repositories."""
+    handler = GithubHandler()
+    async for repo in handler.get_repositories():
+        owner = repo["owner"]["login"]
+        repo_name = repo["name"]
+
+        if not repo_name or repo_name.startswith("-"):
+            logger.error(f"Invalid repository name: {repo_name}")
+            continue
+
+        logger.info(f"Processing repository: {repo_name} owned by {owner}")
+
+        try:
+            existing_hook = await get_webhook_for_repo(handler, owner, repo_name)
+            if existing_hook:
+                if existing_hook.get("active") is False:
+                    logger.info(f"Reactivating webhook for {owner}/{repo_name}")
+                    await delete_webhook_for_repo(handler, owner, repo_name, existing_hook["id"])
+                    await create_webhook_for_repo(handler, owner, repo_name)
+            else:
+                await create_webhook_for_repo(handler, owner, repo_name)
+        except Exception as e:
+            logger.error(f"Error managing webhook for {owner}/{repo_name}: {e}")
+
+async def get_webhook_for_repo(handler: GithubHandler, owner: str, repo: str) -> dict | None:
+    """Check if a webhook exists for a repository."""
+    url = f"{handler.base_url}/repos/{owner}/{repo}/hooks"
+    hooks = await handler.fetch_with_retry(url)
+    app_hook_url = f"{handler.app_host}/integration/hook/github-cloud"
+    for hook in hooks:
+        if hook["config"].get("url") == app_hook_url:
+            logger.info(f"Found existing webhook for {owner}/{repo} with ID {hook['id']}")
+            return hook
+    return None
+
+async def create_webhook_for_repo(handler: GithubHandler, owner: str, repo: str) -> None:
+    """Create a webhook for a repository."""
+    url = f"{handler.base_url}/repos/{owner}/{repo}/hooks"
+    payload = {
+        "name": "web",
+        "active": True,
+        "events": ["push", "pull_request", "issues"],
+        "config": {
+            "url": f"{handler.app_host}/integration/hook/github-cloud",
+            "content_type": "json",
+            "insecure_ssl": "0",
+        },
+    }
+    try:
+        response = await handler.client.post(url, headers=handler.headers, json=payload)
+        if response.status_code == 201:
+            logger.info(f"Webhook created for {owner}/{repo}")
+        else:
+            logger.error(f"Failed to create webhook for {owner}/{repo}: {response.text}")
+    except Exception as e:
+        logger.error(f"Error creating webhook for {owner}/{repo}: {e}")
+
+async def delete_webhook_for_repo(handler: GithubHandler, owner: str, repo: str, hook_id: int) -> None:
+    """Delete a webhook for a repository."""
+    url = f"{handler.base_url}/repos/{owner}/{repo}/hooks/{hook_id}"
+    try:
+        response = await handler.client.delete(url, headers=handler.headers)
+        if response.status_code == 204:
+            logger.info(f"Webhook deleted for {owner}/{repo}")
+        else:
+            logger.error(f"Failed to delete webhook for {owner}/{repo}: {response.text}")
+    except Exception as e:
+        logger.error(f"Error deleting webhook for {owner}/{repo}: {e}")

integrations/github-cloud/github_cloud/webhooks.py

@@ -7,71 +7,84 @@
 
 from port_ocean.context.ocean import ocean
 from client import GithubHandler
+from github_cloud.webhooks import manage_webhooks
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-@ocean.on_resync('repository')
-async def resync_repository(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
-    """Resync repositories."""
-    try:
-        handler = GithubHandler()
-        async for repo in handler.get_repositories():
-            logger.info(f"Yielding repository: {repo['name']}")
-            yield repo
-    except Exception as e:
-        logger.error(f"Failed to resync repository: {e}")
 
-@ocean.on_resync('issue')
-async def resync_issues(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
-    """Resync issues."""
-    try:
-        handler = GithubHandler()
-        async for repo in handler.get_repositories():
-            async for issue in handler.get_issues(repo["owner"]["login"], repo["name"]):
-                logger.info(f"Yielding issue: {issue['title']}")
-                yield issue
-    except Exception as e:
-        logger.error(f"Failed to resync issues: {e}")
+def register_resync_handlers():
+    """Register resync handlers after the PortOcean context is initialized."""
 
-@ocean.on_resync('pull_request')
-async def resync_pull_requests(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
-    """Resync pull requests."""
-    try:
-        handler = GithubHandler()
-        async for repo in handler.get_repositories():
-            async for pull_request in handler.get_pull_requests(repo["owner"]["login"], repo["name"]):
-                logger.info(f"Yielding pull request: {pull_request['title']}")
-                yield pull_request
-    except Exception as e:
-        logger.error(f"Failed to resync pull requests: {e}")
+    @ocean.on_resync('repository')
+    async def resync_repository(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
+        """Resync repositories."""
+        try:
+            handler = GithubHandler()
+            async for repo in handler.get_repositories():
+                logger.info(f"Yielding repository: {repo['name']}")
+                yield repo
+        except Exception as e:
+            logger.error(f"Failed to resync repository: {e}")
 
-@ocean.on_resync('team')
-async def resync_teams(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
-    """Resync teams."""
-    try:
-        handler = GithubHandler()
-        async for org in handler.get_organizations():
-            async for team in handler.get_teams(org["login"]):
-                logger.info(f"Yielding team: {team['name']}")
-                yield team
-    except Exception as e:
-        logger.error(f"Failed to resync teams: {e}")
+    @ocean.on_resync('issue')
+    async def resync_issues(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
+        """Resync issues."""
+        try:
+            handler = GithubHandler()
+            async for repo in handler.get_repositories():
+                async for issue in handler.get_issues(repo["owner"]["login"], repo["name"]):
+                    logger.info(f"Yielding issue: {issue['title']}")
+                    yield issue
+        except Exception as e:
+            logger.error(f"Failed to resync issues: {e}")
 
-@ocean.on_resync('workflow')
-async def resync_workflows(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
-    """Resync workflows."""
-    try:
-        handler = GithubHandler()
-        async for repo in handler.get_repositories():
-            async for workflow in handler.get_workflows(repo["owner"]["login"], repo["name"]):
-                logger.info(f"Yielding workflow: {workflow['name']}")
-                yield workflow
-    except Exception as e:
-        logger.error(f"Failed to resync workflows: {e}")
+    @ocean.on_resync('pull_request')
+    async def resync_pull_requests(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
+        """Resync pull requests."""
+        try:
+            handler = GithubHandler()
+            async for repo in handler.get_repositories():
+                async for pull_request in handler.get_pull_requests(repo["owner"]["login"], repo["name"]):
+                    logger.info(f"Yielding pull request: {pull_request['title']}")
+                    yield pull_request
+        except Exception as e:
+            logger.error(f"Failed to resync pull requests: {e}")
+
+    @ocean.on_resync('team')
+    async def resync_teams(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
+        """Resync teams."""
+        try:
+            handler = GithubHandler()
+            async for org in handler.get_organizations():
+                async for team in handler.get_teams(org["login"]):
+                    logger.info(f"Yielding team: {team['name']}")
+                    yield team
+        except Exception as e:
+            logger.error(f"Failed to resync teams: {e}")
+
+    @ocean.on_resync('workflow')
+    async def resync_workflows(kind: str) -> AsyncGenerator[dict[Any, Any], None]:
+        """Resync workflows."""
+        try:
+            handler = GithubHandler()
+            async for repo in handler.get_repositories():
+                async for workflow in handler.get_workflows(repo["owner"]["login"], repo["name"]):
+                    if "name" in workflow:
+                        logger.info(f"Yielding workflow: {workflow['name']}")
+                        yield workflow
+                    else:
+                        logger.warning(f"Unexpected workflow structure: {workflow}")
+        except Exception as e:
+            logger.error(f"Failed to resync workflows: {e}")
+
+
+# Register resync handlers immediately after the PortOcean context is initialized
+register_resync_handlers()
 
 @ocean.on_start()
 async def on_start() -> None:
     """Handle integration start."""
-    logger.info("Starting GitHub Cloud integration")
+    logger.info("Starting GitHub Cloud integration")
+    await manage_webhooks()