Feat/pbi load test v1 (#1)

README.md

@@ -1,2 +1,66 @@
+![PyPI - Python Version](https://img.shields.io/pypi/pyversions/pbi-load-test?logo=python)
+![PyPI - Version](https://img.shields.io/pypi/v/pbi-load-test?logo=pypi&color=blue&link=https%3A%2F%2Fpypi.org%2Fproject%2Fpbi-load-test%2F)
+
+
+
 # pbi-load-test
 Python package and CLI application to measure Power BI reports loading capacity, using different filters and parameters.
+
+It simulates a realistic set of user actions such as changing slicers, filters (soon), bookmarks (soon)
+
+## Prerequisites
+
+- It requires and Azure AD login method. Ensure you have [**Azure CLI**](https://learn.microsoft.com/fr-fr/cli/azure/install-azure-cli) installed locally and authentificate (`az login`) in order to generate a token easily. Soon:
+
+> [!NOTE]  
+> - Soon: the package will open a Window to authentificate if the Azure login has not be installed
+> - Soon: the package will be able to load Service Principal to to the test (Tenant ID, Client ID and Client Secret)
+
+- This package is based on **Selenium** python package. It will open a Chromium window to launch the test. In any case, it may require the Chromium driver locally. The latest versions for each OS can be found [here](https://chromedriver.chromium.org/downloads).
+
+## Configuration
+
+The load test is configurated through a `config.yaml` file which should be located in the current working directory. 
+
+```yaml
+# authentification: oauth
+workspace: ... # PBI Workspace Name
+report: ... # Report name
+page: ... # Page name
+
+slicers:
+  - table: ... # Table name from dataset which contains the column to filter on
+    column: ... # The column name containing the values to filter on
+    values:
+      - ... # The value to filter on
+      - ...
+```
+
+> [!NOTE]
+> For the moment, only [slicers](https://learn.microsoft.com/en-us/power-bi/visuals/power-bi-visualization-slicers?tabs=powerbi-desktop) are usable to filter on. Later, [filters](https://learn.microsoft.com/en-us/power-bi/create-reports/power-bi-report-add-filter?tabs=powerbi-desktop) will be available.
+> Also, one slicer can be used in this first version. In the future, the tool will be able to iterate between slicers list and create combinations between slicer and filter values.
+> 
+## Example
+
+It ensures that a `config.yaml` file exists in the current working directory
+
+> [!NOTE]  
+> The package will in the future be able to parse `config.yaml` file from different project through the CLI application.
+
+```
+❯ poetry run pbi-load-test run
+
+CORE - MARKETING [DEV]
+SFE Country Dashboard TMDL
+Activity Field Days
+Workspace ID: 310d9fbb-1474-4939-bcb8-669a536ec959
+Report ID: c89485c6-e0c3-4715-a710-ddd450491a9a
+groups/310d9fbb-1474-4939-bcb8-669a536ec959/reports/c89485c6-e0c3-4715-a710-ddd450491a9a/pages
+Page ID: ReportSectioncbd8077dfb6a167ccb5e
+Duration: 30.085
+Press Enter to continue...
+```
+
+<img width="1312" alt="Capture_d’écran_2023-09-04_à_08_32_04_bis" src="https://github.com/lgrosjean/pbi-load-test/assets/34337781/2373d37e-c1c8-4338-8255-b4d0a1dc9284">
+
+After the test, all created files will be removed.

examples/README.md

@@ -0,0 +1,4 @@
+# Examples
+
+- `config.yaml`
+  - Filter on `country name` slicer from `geography` table and `Brazil` value

examples/config.yaml

@@ -0,0 +1,13 @@
+# authentification: oauth
+workspace: ... # PBI Workspace Name
+report: ... # Report name
+page: Activity Field Days
+
+slicers:
+  - table: "geography"
+    column: "country name"
+    values:
+      - "Brazil"
+
+filters:
+# TODO: create jobs list to run

pyproject.toml

@@ -0,0 +1,29 @@
+[tool.poetry]
+name = "pbi-load-test"
+version = "0.1.1a0"
+description = "Python package and CLI application to measure Power BI reports loading capacity, using different filters and parameters."
+authors = ["Léo Grosjean <[email protected]>"]
+readme = "README.md"
+packages = [{ include = "pbi_load_test", from = "src" }]
+
+[tool.poetry.scripts]
+pbi-load-test = "pbi_load_test.cli:cli"
+
+[tool.poetry.dependencies]
+python = "^3.9"
+click = "^8.1.7"
+pyyaml = "^6.0.1"
+requests = "^2.31.0"
+azure-identity = "^1.14.0"
+selenium = "^4.12.0"
+
+[tool.poetry.group.dev]
+optional = true
+
+[tool.poetry.group.dev.dependencies]
+black = "^23.7.0"
+ruff = "^0.0.287"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

src/pbi_load_test/cli.py

@@ -0,0 +1,20 @@
+import click
+
+from .pbi import PowerBIClient
+from .run import run as _run
+
+
+@click.group()
+def cli():
+    """Application to simplify Power BI Load Test."""
+
+
+@cli.command()
+def token():
+    pbi_client = PowerBIClient()
+    click.echo(pbi_client.token, color=True)
+
+
+@cli.command()
+def run():
+    _run()

src/pbi_load_test/config.py

@@ -0,0 +1,18 @@
+"""
+Functions to create a JSON file ready to used by HTML
+"""
+
+import typing as t
+
+import yaml
+
+DEFAULT_CONFIG_PATH = "./config.yaml"
+
+
+# TODO: use pydantic to load config and validate schema
+def load_config(config_path: t.Optional[str] = None):
+    if config_path is None:
+        config_path = DEFAULT_CONFIG_PATH
+
+    with open(config_path, "r", encoding="utf8") as config_file:
+        return yaml.safe_load(config_file)

src/pbi_load_test/pbi.py

@@ -0,0 +1,95 @@
+"""
+ Take Ownership: https://learn.microsoft.com/en-us/rest/api/power-bi/reports/take-over-in-group
+"""
+import typing as t
+from pathlib import Path
+from urllib.parse import urljoin
+
+import requests
+from azure.identity import (  # ClientSecretCredential,; InteractiveBrowserCredential,
+    DefaultAzureCredential,
+)
+
+SCOPE = "https://analysis.windows.net/powerbi/api/.default"
+
+
+class PowerBIClient:
+    base_url = "https://api.powerbi.com/v1.0/myorg/ "
+
+    # TODO: if client_id, tenant_id, client_secret provided, use ClientSecretCredential
+    def __init__(self):
+        # By default: AAD
+        self.token = self.get_token()
+
+    # TODO: check "authentification" method in config: oauth, browser, service principal
+    @staticmethod
+    def get_token():
+        # TODO: check if azure-cli is available, if not use InteractiveBrowser
+        # azure-cli installed (brew install ...)
+        # `az login --allow-no-subscriptions` or see the option with scope in parameters
+        credentials = DefaultAzureCredential()
+        access_token = credentials.get_token(SCOPE)
+        token = access_token.token
+        return token
+
+    def dump_token(self, path: t.Union[str, Path] = "."):
+        token_path = Path(path) / "token.json"
+
+        token_sub_string = '{{"PBIToken":"{0}"}}'.format(self.token)
+        token_str = "accessToken={0};".format(token_sub_string)
+
+        with open(token_path, "w", encoding="utf8") as token_file:
+            token_file.write(token_str)
+
+        return token_path
+
+    @property
+    def headers(self):
+        return {"Authorization": f"Bearer {self.token}"}
+
+    def get(self, endpoint, params: t.Optional[dict] = None):
+        url = urljoin(self.base_url, endpoint)
+        return requests.get(url, params=params, headers=self.headers)
+
+    def get_workspaces(self):
+        # TODO: create dataclass to load json
+        endpoint = "groups"
+        res = self.get(endpoint)
+        return res.json()["value"]
+
+    def get_workspace_id(self, workspace_name: str):
+        workspaces = self.get_workspaces()
+        try:
+            return next(
+                workspace["id"]
+                for workspace in workspaces
+                if workspace["name"] == workspace_name
+            )
+        except StopIteration:
+            return None
+
+    def get_reports(self, workspace_id: str) -> dict:
+        # TODO: create dataclass to load json
+        endpoint = f"groups/{workspace_id}/reports"
+        res = self.get(endpoint)
+        return res.json()["value"]
+
+    def get_report(self, workspace_id: str, report_name: str) -> dict:
+        reports = self.get_reports(workspace_id)
+        return next(report for report in reports if report["name"] == report_name)
+
+    def get_report_id(self, workspace_id: str, report_name: str) -> str:
+        report = self.get_report(workspace_id, report_name)
+        return report["id"]
+
+    def get_pages(self, workspace_id: str, report_id: str) -> list:
+        endpoint = f"groups/{workspace_id}/reports/{report_id}/pages"
+        print(endpoint)
+        res = self.get(endpoint)
+        return res.json()["value"]
+
+    def get_page(self, workspace_id: str, report_id: str, page_page: str) -> dict:
+        pages = self.get_pages(workspace_id, report_id)
+        return next(page for page in pages if page["displayName"] == page_page)
+
+    # https://learn.microsoft.com/en-us/rest/api/power-bi/reports/get-pages-in-group

src/pbi_load_test/run.py

@@ -0,0 +1,98 @@
+import json
+import time
+from pathlib import Path
+
+from selenium.common.exceptions import JavascriptException
+
+from .config import load_config
+from .pbi import PowerBIClient
+from .scraping import copy_html, load_url
+
+
+def run(project_dir=None, clean=True):
+    """Run the load test"""
+
+    if not project_dir:
+        project_dir = Path().cwd()
+
+    pbi_client = PowerBIClient()
+    token_path = pbi_client.dump_token(path=project_dir)
+
+    config = load_config()
+
+    workspace_name = config["workspace"]
+    print(workspace_name)
+    report_name = config["report"]
+    print(report_name)
+    page_name = config.get("page", "")
+    print(page_name)
+
+    # Get Workspace
+    workspace_id = pbi_client.get_workspace_id(workspace_name)
+
+    if not workspace_id:
+        raise Exception
+
+    print(f"Workspace ID: {workspace_id}")
+
+    # Get Report
+    report = pbi_client.get_report(workspace_id, report_name)
+    report_url = report["embedUrl"]
+    report_id = report["id"]
+
+    print(f"Report ID: {report_id}")
+
+    # Get Page
+    page = pbi_client.get_page(workspace_id, report_id, page_name)
+    page_id = page["name"]
+
+    print(f"Page ID: {page_id}")
+
+    # Fitlers
+    slicers = config.get("slicers", [])
+    filters = config.get("filters", [])
+
+    report_path = project_dir / "parameters.json"
+
+    report_dict = {
+        "reportUrl": report_url,
+        "reportId": report_id,
+        "pageName": page_id,
+        "sessionRestart": 100,
+        "slicers": slicers,
+        "filters": filters,
+    }
+
+    report_parameter_str = json.dumps(report_dict)
+
+    report_str = "reportParameters={0};".format(report_parameter_str)
+
+    with open(report_path, "w", encoding="utf8") as report_file:
+        report_file.write(report_str)
+
+    project_filepath = copy_html(project_dir).absolute()
+
+    fileurl = f"file:///{project_filepath}"
+
+    driver = load_url(fileurl)
+
+    duration = None
+
+    while not duration:
+        try:
+            duration = driver.execute_script("return duration")
+
+        except JavascriptException:
+            time.sleep(1)
+
+    print(f"Duration: {duration}s")
+
+    input("Press Enter to continue...")
+
+    driver.quit()
+
+    if clean:
+        print("Start cleaning...")
+        Path(report_path).unlink()
+        Path(token_path).unlink()
+        Path(project_filepath).unlink()

src/pbi_load_test/scraping.py

@@ -0,0 +1,25 @@
+from pathlib import Path
+from shutil import copy
+
+from selenium import webdriver
+from selenium.webdriver.chrome.options import Options
+
+HTML_FILENAME = "LoadTest.html"
+HTML_FILEPATH = Path(__file__).parent / "static" / HTML_FILENAME
+
+
+def copy_html(dst_dir) -> Path:
+    dst_path = Path(dst_dir) / HTML_FILENAME
+    copy(HTML_FILEPATH, str(dst_path))
+    return dst_path
+
+
+def load_url(fileurl: str):
+    chrome_options = Options()
+    # chrome_options.add_argument("--incognito")
+    chrome_options.add_experimental_option("detach", True)
+    driver = webdriver.Chrome(options=chrome_options)
+
+    driver.get(fileurl)
+
+    return driver