Skip to content

Testing OIDC

Testing OIDC #278

---
name: Integration tests on Fabric DW
on: # yamllint disable-line rule:truthy
workflow_dispatch:
pull_request:
branches:
- oidc_connect
permissions:
id-token: write
contents: read
jobs:
integration-tests-fabric-dw:
name: Regular
strategy:
fail-fast: false
max-parallel: 1
matrix:
profile: ["ci_azure_auto"]
python_version: ["3.11"]
msodbc_version: ["17", "18"]
runs-on: ubuntu-latest
steps:
# Azure login using federated credentials
- name: Azure login with OIDC
uses: azure/login@v2
with:
client-id: ${{ secrets.DBT_AZURE_SP_NAME }}
tenant-id: ${{ secrets.DBT_AZURE_TENANT }}
allow-no-subscriptions: true
federated-token: true
- name: Install ODBC Driver 18 for SQL Server (Ubuntu)
run: |
sudo apt-get update
sudo apt-get install -y gnupg
curl https://packages.microsoft.com/keys/microsoft.asc | sudo apt-key add -
curl https://packages.microsoft.com/config/ubuntu/$(lsb_release -rs)/prod.list | sudo tee /etc/apt/sources.list.d/mssql-release.list
sudo apt-get update
sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18
- name: Connect to Azure SQL Database
run: |
python3 -m venv venv
source venv/bin/activate
pip install azure-identity pyodbc azure-core
python - <<EOF
from azure.core.credentials import AccessToken
from azure.identity import DefaultAzureCredential
import pyodbc
import logging
try:
credential = DefaultAzureCredential()
token = credential.get_token("https://database.windows.net/.default")
connection_string = (
"Driver={ODBC Driver 18 for SQL Server};"
"Server=x6eps4xrq2xudenlfv6naeo3i4-6xw4uystlgdevluyqmndlcagwe.msit-datawarehouse.fabric.microsoft.com;"
"Database=permissionstest;"
"Authentication=ActiveDirectoryAccessToken;"
)
access_token = token.token
print(connection_string)
logging.info("connection strin is: ", connection_string)
connection = pyodbc.connect(connection_string, attrs_before={1256: access_token})
cursor = connection.cursor()
cursor.execute("SELECT TOP 10 * FROM dbo.Trip")
rows = cursor.fetchall()
for row in rows:
print(row)
connection.close()
except pyodbc.Error as e:
logging.error("Error occurred while connecting to the database.", exc_info=True)
EOF