Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 10 additions & 8 deletions msa_sdk/msa_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,15 +4,17 @@
import os
import random
import sys
from typing import Optional

import requests
from requests import Response

from msa_sdk import constants
from msa_sdk import context

logger = logging.getLogger("msa-sdk")

def host_port():
def host_port() -> tuple[str, str]:
"""
Hostname and port of the API.

Expand All @@ -37,11 +39,11 @@ class MSA_API(): # pylint: disable=invalid-name

def __init__(self):
"""Initialize."""
self.url = 'http://{}:{}/ubi-api-rest'.format(*host_port())
self.path = ""
self.response = None
self.log_response = True
self._content = ""
self.url: str = 'http://{}:{}/ubi-api-rest'.format(*host_port())
self.path: str = ""
self.response: Optional[Response] = None
self.log_response: bool = True
self._content: str = ""
self.action = self.__class__

@classmethod
Expand Down Expand Up @@ -224,7 +226,7 @@ def check_response(self):


"""
if not self.response.ok:
if self.response is not None and not self.response.ok:
json_response = self.response.json()
self._content = self.process_content(self.FAILED, self.action,
json_response['message'])
Expand Down Expand Up @@ -329,7 +331,7 @@ def _call_delete(self) -> None:
self._content = self.response.text
self.check_response()

def add_trace_headers(self, headers):
def add_trace_headers(self, headers: dict[str, str]):
"""Add W3C trace headers."""
if 'TRACEID' not in context:
t, s = self.create_trace_id()
Expand Down
17 changes: 11 additions & 6 deletions msa_sdk/profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,7 @@
The Profile class inherits from MSA_API and provides methods to check the existence
of profiles and perform other profile-related operations.
"""

import json
from urllib.parse import urlencode

from msa_sdk.msa_api import MSA_API

Expand Down Expand Up @@ -41,7 +40,13 @@ def exist(self, reference) -> bool:
True if the profile exists, False otherwise.
"""
self.action = 'Check Profile exist by reference'
self.path = '{}/v1/exist/{}'.format(self.api_path, reference)
self._call_post()
result = json.loads(self.content)
return result.get('exist', False)
url_encoded = urlencode({'extRef': reference})
self.path = '{}/ref?{}'.format(self.api_path, url_encoded)
self._call_get()
if self.response is None:
raise Exception("No response received from the server.")
if self.response.status_code == 404:
return False
if self.response.status_code == 200:
return True
raise Exception("Unexpected response code: {}".format(self.response.status_code))
20 changes: 0 additions & 20 deletions tests/test_profie.py

This file was deleted.

45 changes: 45 additions & 0 deletions tests/test_profile_exist.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
import pytest

from msa_sdk.profile import Profile


class DummyResponse:
def __init__(self, status_code):
self.status_code = status_code

class DummyProfile(Profile):
def _call_get(self):
# This will be set in the test
pass

def test_exist_returns_true(monkeypatch):
profile = DummyProfile()
def fake_call_get():
profile.response = DummyResponse(200)
profile._call_get = fake_call_get
assert profile.exist('ref1') is True

def test_exist_returns_false(monkeypatch):
profile = DummyProfile()
def fake_call_get():
profile.response = DummyResponse(404)
profile._call_get = fake_call_get
assert profile.exist('ref2') is False


def test_exist_raises_no_response(monkeypatch):
profile = DummyProfile()
def fake_call_get():
profile.response = None
profile._call_get = fake_call_get
with pytest.raises(Exception, match="No response received from the server."):
profile.exist('ref3')


def test_exist_raises_unexpected_code(monkeypatch):
profile = DummyProfile()
def fake_call_get():
profile.response = DummyResponse(500)
profile._call_get = fake_call_get
with pytest.raises(Exception, match="Unexpected response code: 500"):
profile.exist('ref4')