Skip to content

Commit

Permalink
remove default basic auth from integration test
Browse files Browse the repository at this point in the history
  • Loading branch information
@ipamaas
ipamaas committed Jul 6, 2023
1 parent 01dde3a commit ffdc3da
Showing 1 changed file with 64 additions and 46 deletions.
110 changes: 64 additions & 46 deletions test/test_integration.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,8 @@


# Check env variables
api_username = os.getenv('API_USERNAME', 'admin')
api_password = os.getenv('API_PASSWORD', 'password')
api_username = os.getenv('API_USERNAME')
api_password = os.getenv('API_PASSWORD')
customIp = None
if "TESTENV_CUSTOMIP" in os.environ:
customIp = os.environ["TESTENV_CUSTOMIP"]
Expand Down Expand Up @@ -103,22 +103,26 @@ class IntegrationTestMSBClientRestInterfaces(unittest.TestCase):
def test_smartobjectmanagement_availability(self):
logging.info("Smart Object URL: >" + so_url + "<")

response = requests.get(
so_url + "/service/token/" + owner_uuid,
verify=False,
auth=HTTPBasicAuth(api_username, api_password)
)
params = {
"url": so_url + "/service/token/" + owner_uuid,
"verify": False,
}
if api_username and api_password:
params["auth"] = HTTPBasicAuth(api_username, api_password)
response = requests.get(**params)
self.assertEqual(response.status_code, 201, "Can not reach smart-object-management")

@pytest.mark.run(order=2)
def test_integrationflowmanagement_availability(self):
logging.info("Flow URL: >" + str(flow_url) + "<")

response = requests.get(
flow_url + "/integrationFlow/customer/" + owner_uuid,
verify=False,
auth=HTTPBasicAuth(api_username, api_password)
)
params = {
"url": flow_url + "/integrationFlow/customer/" + owner_uuid,
"verify": False,
}
if api_username and api_password:
params["auth"] = HTTPBasicAuth(api_username, api_password)
response = requests.get(**params)
self.assertEqual(response.status_code, 200, "Can not reach integration-flow-management")


Expand All @@ -130,11 +134,13 @@ def test_getVerificationToken(self):
logging.info("Broker URL: >" + str(broker_url) + "<")

# get valid verification token
response = requests.get(
so_url + "/service/token/" + owner_uuid,
verify=False,
auth=HTTPBasicAuth(api_username, api_password)
)
params = {
"url": so_url + "/service/token/" + owner_uuid,
"verify": False,
}
if api_username and api_password:
params["auth"] = HTTPBasicAuth(api_username, api_password)
response = requests.get(**params)
self.assertEqual(
response.status_code,
201,
Expand Down Expand Up @@ -245,16 +251,18 @@ def test_communication(self):
logging.debug(flow_json)

# create flow
response = requests.post(
flow_url + "/integrationFlow/create",
headers={
params = {
"url": flow_url + "/integrationFlow/create",
"headers": {
"Accept": "application/json",
"Content-Type": "application/json",
},
data=str(flow_json),
verify=False,
auth=HTTPBasicAuth(api_username, api_password)
)
"data": str(flow_json),
"verify": False,
}
if api_username and api_password:
params["auth"] = HTTPBasicAuth(api_username, api_password)
response = requests.post(**params)
response_dict = json.loads(response.text)
self.assertEqual(
response.status_code, 201, str("Creating integration flow failed: " + str(response))
Expand All @@ -274,11 +282,13 @@ def test_communication(self):
# Test deploy integration flow
#
# deploy flow
response = requests.put(
flow_url + "/integrationFlow/" + str(flow_id) + "/deploy",
verify=False,
auth=HTTPBasicAuth(api_username, api_password)
)
params = {
"url": flow_url + "/integrationFlow/" + str(flow_id) + "/deploy",
"verify": False,
}
if api_username and api_password:
params["auth"] = HTTPBasicAuth(api_username, api_password)
response = requests.put(**params)
self.assertEqual(
response.status_code, 200, str("Deploying integration flow failed: " + str(response))
)
Expand Down Expand Up @@ -361,11 +371,13 @@ def test_communication(self):
#
# undeploy flow
if flowDeployed:
response = requests.put(
flow_url + "/integrationFlow/" + str(flow_id) + "/undeploy",
verify=False,
auth=HTTPBasicAuth(api_username, api_password)
)
params = {
"url": flow_url + "/integrationFlow/" + str(flow_id) + "/undeploy",
"verify": False,
}
if api_username and api_password:
params["auth"] = HTTPBasicAuth(api_username, api_password)
response = requests.put(**params)
self.assertEqual(
response.status_code,
200,
Expand All @@ -379,15 +391,17 @@ def test_communication(self):
#
# delete flow
if flowCreated:
response = requests.delete(
flow_url + "/integrationFlow/" + str(flow_id),
headers={
params = {
"url": flow_url + "/integrationFlow/" + str(flow_id),
"headers": {
"Accept": "application/json",
"Content-Type": "application/json",
},
verify=False,
auth=HTTPBasicAuth(api_username, api_password)
)
"verify": False,
}
if api_username and api_password:
params["auth"] = HTTPBasicAuth(api_username, api_password)
response = requests.delete(**params)
self.assertEqual(
response.status_code, 200, str("Deleting integration flow failed")
)
Expand All @@ -399,12 +413,16 @@ def test_communication(self):
#
# delete smart object
if soCreated:
response = requests.delete(
so_url + "/smartobject/" + SO_UUID,
headers={"Content-Type": "application/json"},
verify=False,
auth=HTTPBasicAuth(api_username, api_password)
)
params = {
"url": so_url + "/smartobject/" + SO_UUID,
"headers": {
"Content-Type": "application/json",
},
"verify": False,
}
if api_username and api_password:
params["auth"] = HTTPBasicAuth(api_username, api_password)
response = requests.delete(**params)
self.assertEqual(response.status_code, 200, "Problem deleting smart object")

time.sleep(WAITING_TIME)
Expand Down

0 comments on commit ffdc3da

Please sign in to comment.