Merge pull request #1 from lgrosjean/feat/add-tests

Feat/add tests

.github/workflows/check.yml

@@ -0,0 +1,84 @@
+name: Check
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+  workflow_dispatch:
+
+env:
+  PACKAGE_DIR: powerbi_ext
+  TESTS_DIR: tests
+
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.10", "3.11"]
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install Poetry
+        run: |
+          pipx install poetry
+          poetry --version
+
+      - name: Install dependencies
+        run: |
+          poetry install
+
+      - name: Test with pytest
+        run: poetry run pytest --cov=$PACKAGE_DIR $TESTS_DIR
+
+  lint:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.10", "3.11"]
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install Poetry
+        run: |
+          pipx install poetry
+          poetry --version
+
+      - name: Install dependencies
+        run: |
+          poetry install
+
+      - name: Lint with pylint
+        run: poetry run pylint --jobs 0 $PACKAGE_DIR --fail-under 9
+
+  format:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.10", "3.11"]
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install black
+        run: pip install black
+
+      - name: Check format with black
+        run: black --check  $PACKAGE_DIR

powerbi_ext/auth.py

@@ -1,15 +1,18 @@
+"""PowerBI authentication module."""
 import os
+import typing as t
 
 from azure.identity import ClientSecretCredential
 
 SCOPE = "https://analysis.windows.net/powerbi/api/.default"
 
 
-def get_token(
-    tenant_id: str = None,
-    client_id: str = None,
-    client_secret: str = None,
+def get_credential(
+    tenant_id: t.Optional[str] = None,
+    client_id: t.Optional[str] = None,
+    client_secret: t.Optional[str] = None,
 ):
+    """Get Azure ClientSecretCredential using Meltano env variables"""
     if not tenant_id:
         tenant_id = os.environ["POWERBI_EXT_TENANT_ID"]
     if not client_id:
@@ -23,4 +26,21 @@ def get_token(
         client_secret=client_secret,
     )
 
-    return credential.get_token(SCOPE).token
+    return credential
+
+
+def get_token(
+    tenant_id: t.Optional[str] = None,
+    client_id: t.Optional[str] = None,
+    client_secret: t.Optional[str] = None,
+):
+    """Get Azure token"""
+    credential = get_credential(
+        tenant_id=tenant_id,
+        client_id=client_id,
+        client_secret=client_secret,
+    )
+
+    access_token = credential.get_token(SCOPE)
+    token = access_token.token
+    return token

powerbi_ext/extension.py

@@ -17,10 +17,11 @@
 class PowerBIExtension(ExtensionBase):
     """Extension implementing the ExtensionBase interface."""
 
-    def __init__(self) -> None:
+    def __init__(self, token: t.Optional[str] = None) -> None:
         """Initialize the extension."""
         self.log = structlog.get_logger(name=self.__class__.__name__)
-        token = get_token()
+        if not token:
+            token = get_token()
         self.log.info("Bearer token accessed.")
         self.headers = {"Authorization": f"Bearer {token}"}
 
@@ -57,9 +58,9 @@ def refresh(
         res = requests.post(url, json=body, headers=self.headers, timeout=TIMEOUT)
         self.log.info(res.status_code)
         if res.status_code != 200:
-            print(res.reason, res.headers)
-        else:
-            return res.headers["RequestId"]
+            self.log.error(res.reason, res.headers)
+            raise requests.RequestException(res.status_code, res.reason, res.headers)
+        return res.headers["RequestId"]
 
     def describe(self) -> models.Describe:
         """Describe the extension.

powerbi_ext/main.py

@@ -1,7 +1,5 @@
 """PowerBI cli entrypoint."""
 
-import sys
-
 import structlog
 import typer
 from meltano.edk.extension import DescribeFormat
@@ -13,8 +11,6 @@
 
 log = structlog.get_logger(APP_NAME)
 
-ext = PowerBIExtension()
-
 app = typer.Typer(
     name=APP_NAME,
     pretty_exceptions_enable=False,
@@ -28,13 +24,8 @@ def describe(
     )
 ) -> None:
     """Describe the available commands of this extension."""
-    try:
-        typer.echo(ext.describe_formatted(output_format))
-    except Exception:
-        log.exception(
-            "describe failed with uncaught exception, please report to maintainer"
-        )
-        sys.exit(1)
+    ext = PowerBIExtension()
+    typer.echo(ext.describe_formatted(output_format))
 
 
 @app.command()
@@ -43,14 +34,12 @@ def refresh(
         None,
         "-w",
         "--workspace",
-        envvar="POWERBI_EXT_WORKSPACE_ID",
-        show_envvar=True,
         help="Workspace ID. If not provided, will look for Dataset in 'My Workspace'",
     ),
     dataset_id: str = typer.Argument(..., help="Dataset ID"),
 ) -> None:
     """Refresh the given dataset in the given workspace"""
-
+    ext = PowerBIExtension()
     typer.echo(ext.refresh(workspace_id, dataset_id))
 
 

pyproject.toml

@@ -24,10 +24,13 @@ typer = "^0.6.1"
 azure-identity = "^1.14.0"
 requests = "^2.31.0"
 
-[tool.poetry.dev-dependencies]
+[tool.poetry.group.dev.dependencies]
 black = "^22.3.0"
 isort = "^5.10.1"
 flake8 = "^3.9.0"
+pytest = "^7.4.2"
+pytest-cov = "^4.1.0"
+pylint = "^3.0.1"
 
 [build-system]
 requires = ["poetry-core>=1.0.8"]

tests/test_auth.py

@@ -0,0 +1,79 @@
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+from azure.identity import ClientSecretCredential
+
+from powerbi_ext.auth import SCOPE, get_credential, get_token
+
+TOKEN = "token"
+TENANT_ID, CLIENT_ID, CLIENT_SECRET = "tenant_id", "client_id", "client_secret"
+
+
+def test_get_credential_with_args():
+    with patch.object(ClientSecretCredential, "__new__") as mock_ClientSecretCredential:
+        get_credential(
+            tenant_id=TENANT_ID,
+            client_id=CLIENT_ID,
+            client_secret=CLIENT_SECRET,
+        )
+
+        mock_ClientSecretCredential.assert_called_once_with(
+            ClientSecretCredential,
+            tenant_id=TENANT_ID,
+            client_id=CLIENT_ID,
+            client_secret=CLIENT_SECRET,
+        )
+
+
+def test_get_credential_without_args_missing_envvar_tenant_id():
+    if os.getenv("POWERBI_EXT_TENANT_ID"):
+        del os.environ["POWERBI_EXT_TENANT_ID"]
+    os.environ["POWERBI_EXT_CLIENT_ID"] = CLIENT_ID
+    os.environ["POWERBI_EXT_CLIENT_SECRET"] = CLIENT_SECRET
+
+    with pytest.raises(KeyError, match="POWERBI_EXT_TENANT_ID"):
+        get_credential()
+
+
+def test_get_credential_without_args_missing_envvar_client_id():
+    os.environ["POWERBI_EXT_TENANT_ID"] = TENANT_ID
+    if os.getenv("POWERBI_EXT_CLIENT_ID"):
+        del os.environ["POWERBI_EXT_CLIENT_ID"]
+    os.environ["POWERBI_EXT_CLIENT_SECRET"] = CLIENT_SECRET
+
+    with pytest.raises(KeyError, match="POWERBI_EXT_CLIENT_ID"):
+        get_credential()
+
+
+def test_get_credential_without_args_missing_envvar_client_secret():
+    os.environ["POWERBI_EXT_TENANT_ID"] = TENANT_ID
+    os.environ["POWERBI_EXT_CLIENT_ID"] = CLIENT_ID
+    if os.getenv("POWERBI_EXT_CLIENT_SECRET"):
+        del os.environ["POWERBI_EXT_CLIENT_SECRET"]
+
+    with pytest.raises(KeyError, match="POWERBI_EXT_CLIENT_SECRET"):
+        get_credential()
+
+
+def test_get_token():
+    mock_access_token = MagicMock(token=TOKEN)
+    mock_get_token = MagicMock(return_value=mock_access_token)
+    mock_credential = MagicMock(get_token=mock_get_token)
+    with patch(
+        "powerbi_ext.auth.get_credential", return_value=mock_credential
+    ) as mock_get_credential:
+        result = get_token(
+            tenant_id=TENANT_ID,
+            client_id=CLIENT_ID,
+            client_secret=CLIENT_SECRET,
+        )
+
+        mock_get_credential.assert_called_once_with(
+            tenant_id=TENANT_ID,
+            client_id=CLIENT_ID,
+            client_secret=CLIENT_SECRET,
+        )
+        mock_get_token.assert_called_once_with(SCOPE)
+
+        assert result == TOKEN

tests/test_extension.py

@@ -0,0 +1,75 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+from meltano.edk.models import Describe, ExtensionCommand
+from requests import RequestException
+
+from powerbi_ext.extension import BASE_URL, TIMEOUT, PowerBIExtension
+
+TOKEN = "token"
+WORKSPACE_ID = "workspace_id"
+DATASET_ID = "dataset_id"
+
+
+@patch("powerbi_ext.extension.get_token", return_value=TOKEN)
+def test_init_not_token(mock_get_token: MagicMock):
+    ext = PowerBIExtension()
+    mock_get_token.assert_called_once()
+    assert ext.log
+    assert ext.headers == {"Authorization": f"Bearer {TOKEN}"}
+
+
+class TestExtension:
+    ext = PowerBIExtension(token=TOKEN)
+
+    def test_invoke(self):
+        with pytest.raises(NotImplementedError):
+            self.ext.invoke()
+
+    @patch("requests.post")
+    def test_refresh_ok(self, mock_post: MagicMock):
+        mock_res = MagicMock(status_code=200, headers={"RequestId": "RequestId"})
+        url = BASE_URL + f"/groups/{WORKSPACE_ID}/datasets/{DATASET_ID}" + "/refreshes"
+        body = {
+            "notifyOption": "MailOnCompletion",
+        }
+        mock_post.return_value = mock_res
+        res = self.ext.refresh(workspace_id=WORKSPACE_ID, dataset_id=DATASET_ID)
+        mock_post.assert_called_once_with(
+            url, json=body, headers=self.ext.headers, timeout=TIMEOUT
+        )
+        assert res == "RequestId"
+
+    @patch("requests.post")
+    def test_refresh_not_ok(self, mock_post: MagicMock):
+        mock_res = MagicMock(status_code=202)
+        url = BASE_URL + f"/groups/{WORKSPACE_ID}/datasets/{DATASET_ID}" + "/refreshes"
+        body = {
+            "notifyOption": "MailOnCompletion",
+        }
+        mock_post.return_value = mock_res
+        with pytest.raises(RequestException):
+            self.ext.refresh(workspace_id=WORKSPACE_ID, dataset_id=DATASET_ID)
+
+        mock_post.assert_called_once_with(
+            url, json=body, headers=self.ext.headers, timeout=TIMEOUT
+        )
+
+    @patch.object(ExtensionCommand, "__new__")
+    @patch.object(Describe, "__new__")
+    def test_describe(
+        self,
+        mock_describe_class: MagicMock,
+        mock_command_class: MagicMock,
+    ):
+        name, description = "powerbi_extension", "extension commands"
+        mock_command = MagicMock(name=name, description=description)
+        mock_command_class.return_value = mock_command
+
+        self.ext.describe()
+
+        mock_command_class.assert_called_once_with(
+            ExtensionCommand, name=name, description=description
+        )
+
+        mock_describe_class.assert_called_once_with(Describe, commands=[mock_command])