Skip to content

Testing OIDC

Testing OIDC #289

---
name: Integration tests on Fabric DW
on: # yamllint disable-line rule:truthy
workflow_dispatch:
pull_request:
branches:
- oidc_connect
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
permissions:
contents: read # Required to access repository files
packages: read # Grant explicit read access to packages
id-token: write # Needed if using OIDC authentication
container:
image: ghcr.io/${{ github.repository }}:CI-${{ matrix.python_version }}-msodbc${{ matrix.msodbc_version }}
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: Connect to Azure SQL Database
run: |
pip install azure-identity pyodbc azure-core
python - <<EOF
from azure.core.credentials import AccessToken
from azure.identity import DefaultAzureCredential
import pyodbc
import logging
import struct
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;"
)
access_token = token.token.encode('utf-16-le')
token_struct = struct.pack(f'<I{len(access_token)}s', len(access_token), access_token)
SQL_COPT_SS_ACCESS_TOKEN = 1256 # This connection option is defined by microsoft in msodbcsql.h
connection = pyodbc.connect(connection_string, attrs_before={SQL_COPT_SS_ACCESS_TOKEN: token_struct})
logging.info("connection strin is: ", token_struct)
print("token struct is ", token_struct)
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