Compare commits
2 Commits
5412c3ea09
...
e6bca1ce47
| Author | SHA1 | Date | |
|---|---|---|---|
| e6bca1ce47 | |||
| 2addc85e40 |
1
devops/__init__.py
Normal file
1
devops/__init__.py
Normal file
@@ -0,0 +1 @@
|
|||||||
|
# devops package
|
||||||
@@ -59,9 +59,9 @@ def get_token(
|
|||||||
|
|
||||||
def secret_credentials_auth(
|
def secret_credentials_auth(
|
||||||
scope: str = DEVOPS_SCOPE,
|
scope: str = DEVOPS_SCOPE,
|
||||||
tenant_id: str = os.environ.get("AZURE_TENANT_ID", ""),
|
tenant_id = os.environ.get("AZURE_TENANT_ID", ""),
|
||||||
client_id: str = os.environ.get("AZURE_CLIENT_ID", ""),
|
client_id = os.environ.get("AZURE_CLIENT_ID", ""),
|
||||||
client_secret: str = os.environ.get("AZURE_CLIENT_SECRET")
|
client_secret = os.environ.get("AZURE_CLIENT_SECRET")
|
||||||
) -> str:
|
) -> str:
|
||||||
"""
|
"""
|
||||||
Authenticate using client credentials. Pass credentials via environment variables,
|
Authenticate using client credentials. Pass credentials via environment variables,
|
||||||
@@ -92,8 +92,8 @@ def certificate_credentials_auth(
|
|||||||
# Wczytaj klucz prywatny (RSA)
|
# Wczytaj klucz prywatny (RSA)
|
||||||
with open(pem_path, "rb") as f:
|
with open(pem_path, "rb") as f:
|
||||||
pem = f.read()
|
pem = f.read()
|
||||||
key_pem = re.search(b"-----BEGIN (?:RSA )?PRIVATE KEY-----.*?END (?:RSA )?PRIVATE KEY-----", pem, re.S).group(0)
|
key_pem = re.search(b"-----BEGIN (?:RSA )?PRIVATE KEY-----.*?END (?:RSA )?PRIVATE KEY-----", pem, re.S).group(0) # type: ignore
|
||||||
cert_pem = re.search(b"-----BEGIN CERTIFICATE-----.*?END CERTIFICATE-----", pem, re.S).group(0)
|
cert_pem = re.search(b"-----BEGIN CERTIFICATE-----.*?END CERTIFICATE-----", pem, re.S).group(0) # type: ignore
|
||||||
|
|
||||||
private_key = serialization.load_pem_private_key(key_pem, password=None)
|
private_key = serialization.load_pem_private_key(key_pem, password=None)
|
||||||
cert = x509.load_pem_x509_certificate(cert_pem)
|
cert = x509.load_pem_x509_certificate(cert_pem)
|
||||||
@@ -115,7 +115,7 @@ def certificate_credentials_auth(
|
|||||||
|
|
||||||
headers = {"x5t": x5t, "kid": x5t}
|
headers = {"x5t": x5t, "kid": x5t}
|
||||||
|
|
||||||
assertion = jwt.encode(claims, private_key, algorithm="RS256", headers=headers)
|
assertion = jwt.encode(claims, private_key, algorithm="RS256", headers=headers) # type: ignore
|
||||||
|
|
||||||
data = {
|
data = {
|
||||||
"grant_type": "client_credentials",
|
"grant_type": "client_credentials",
|
||||||
@@ -4,15 +4,11 @@ import requests
|
|||||||
import urllib.parse
|
import urllib.parse
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
import logging
|
import logging
|
||||||
from sk.logger import log_entity_creation
|
|
||||||
|
|
||||||
DEVOPS_SCOPE = "https://app.vssps.visualstudio.com/.default"
|
DEVOPS_SCOPE = "https://app.vssps.visualstudio.com/.default"
|
||||||
DEVOPS_API_VERSION = "7.1"
|
DEVOPS_API_VERSION = "7.1"
|
||||||
|
|
||||||
# Get logger. It should be configured by the main application.
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
# log.setLevel(logging.DEBUG)
|
|
||||||
# log.propagate = False
|
|
||||||
|
|
||||||
# Define a class decorator
|
# Define a class decorator
|
||||||
def auto_properties(mapping: dict[str,str]):
|
def auto_properties(mapping: dict[str,str]):
|
||||||
@@ -83,7 +79,6 @@ class Organization():
|
|||||||
self._org_url = org_url.rstrip("/") + "/" # Ensure trailing slash
|
self._org_url = org_url.rstrip("/") + "/" # Ensure trailing slash
|
||||||
self._token = token
|
self._token = token
|
||||||
self._api_version = api_version
|
self._api_version = api_version
|
||||||
log_entity_creation(log, Organization, self._org_url)
|
|
||||||
|
|
||||||
def get_path(self, path: str, params: dict = {}) -> requests.Response:
|
def get_path(self, path: str, params: dict = {}) -> requests.Response:
|
||||||
return get_url(
|
return get_url(
|
||||||
@@ -131,8 +126,6 @@ class Project():
|
|||||||
except ValueError:
|
except ValueError:
|
||||||
raise ValueError(f"Invalid project ID: {self._id}")
|
raise ValueError(f"Invalid project ID: {self._id}")
|
||||||
|
|
||||||
log_entity_creation(log, Project, self.id)
|
|
||||||
|
|
||||||
def get_auto_properties(self):
|
def get_auto_properties(self):
|
||||||
r = get_url(
|
r = get_url(
|
||||||
URL=f"{self._org._org_url}_apis/projects/{self._id}",
|
URL=f"{self._org._org_url}_apis/projects/{self._id}",
|
||||||
@@ -195,7 +188,6 @@ class Repository():
|
|||||||
|
|
||||||
# set other properties if provided
|
# set other properties if provided
|
||||||
self.from_args(**kwargs) # type: ignore[attr-defined]
|
self.from_args(**kwargs) # type: ignore[attr-defined]
|
||||||
log_entity_creation(log, Repository, self.id)
|
|
||||||
|
|
||||||
def get_auto_properties(self):
|
def get_auto_properties(self):
|
||||||
id = self._id if hasattr(self, "_id") else self._name # type: ignore[attr-defined]
|
id = self._id if hasattr(self, "_id") else self._name # type: ignore[attr-defined]
|
||||||
@@ -245,7 +237,6 @@ class Item():
|
|||||||
self.from_args(**kwargs) # type: ignore[attr-defined]
|
self.from_args(**kwargs) # type: ignore[attr-defined]
|
||||||
if "branch" in kwargs:
|
if "branch" in kwargs:
|
||||||
self._branch = kwargs.get("branch")
|
self._branch = kwargs.get("branch")
|
||||||
log_entity_creation(log, Item, self.path)
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def branch(self):
|
def branch(self):
|
||||||
@@ -1,8 +1,8 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
from sk.devops import Organization, Project, Repository, Item
|
from devops.devops import Organization, Project, Repository, Item
|
||||||
from sk.azure import get_token
|
from devops.azure import get_token
|
||||||
|
|
||||||
org = Organization("https://dev.azure.com/mcovsandbox", token=get_token())
|
org = Organization("https://dev.azure.com/mcovsandbox", token=get_token())
|
||||||
|
|
||||||
|
|||||||
4
tests.py
4
tests.py
@@ -1,8 +1,8 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
import unittest
|
import unittest
|
||||||
import requests
|
import requests
|
||||||
from sk.devops import Organization, Repository, Project, Item
|
from devops.devops import Organization, Repository, Project, Item
|
||||||
from sk.azure import get_token
|
from devops.azure import get_token
|
||||||
|
|
||||||
# Get the token outside the test class to speed up tests.
|
# Get the token outside the test class to speed up tests.
|
||||||
# Each Unit test instantinates the class, so doing it here avoids repeated authentication.
|
# Each Unit test instantinates the class, so doing it here avoids repeated authentication.
|
||||||
|
|||||||
Reference in New Issue
Block a user