Skip to content

Commit

Permalink
Commented out bulk aggregation tests for now
Browse files Browse the repository at this point in the history
  • Loading branch information
roywilly committed Nov 1, 2024
1 parent 92c142f commit 126b6c5
Show file tree
Hide file tree
Showing 5 changed files with 129 additions and 128 deletions.
45 changes: 23 additions & 22 deletions tests/test_access/tst_access_drogon_affiliate_login.py
Original file line number Diff line number Diff line change
Expand Up @@ -204,28 +204,29 @@ def test_read_restricted_classification_data(explorer: Explorer):
print("Hits on restricted:", hits)
assert hits >= 1

@pytest.mark.skipif(not (sys.platform == "linux" and
sys.version_info[:2] == (3, 11)),
reason="Test only on single platform/version.")
def test_aggregate_bulk(explorer: Explorer):
"""Test a bulk aggregation method"""
print("Running test:", inspect.currentframe().f_code.co_name)
# Fixed test case ("Drogon_AHM_2023-02-22") in Sumo/DEV
# This user does not have WRITE and hence this should fail
# (In addition this case is not set up with affiliate access)
TESTCASE_UUID = "10f41041-2c17-4374-a735-bb0de62e29dc"
print("About to trigger bulk aggregation on case", TESTCASE_UUID)
body = {
"operations": ["min"],
"case_uuid": TESTCASE_UUID,
"class": "surface",
"iteration_name": "iter-0",
}
with pytest.raises(Exception, match="40*"):
response = explorer._sumo.post(f"/aggregations", json=body)
print("Execution should never reach this line")
print("Unexpected status: ", response.status_code)
print("Unexpected response: ", response.text)
# Remove or update this test when bulk aggregation is finalized
# @pytest.mark.skipif(not (sys.platform == "linux" and
# sys.version_info[:2] == (3, 11)),
# reason="Test only on single platform/version.")
# def test_aggregate_bulk(explorer: Explorer):
# """Test a bulk aggregation method"""
# print("Running test:", inspect.currentframe().f_code.co_name)
# # Fixed test case ("Drogon_AHM_2023-02-22") in Sumo/DEV
# # This user does not have WRITE and hence this should fail
# # (In addition this case is not set up with affiliate access)
# TESTCASE_UUID = "10f41041-2c17-4374-a735-bb0de62e29dc"
# print("About to trigger bulk aggregation on case", TESTCASE_UUID)
# body = {
# "operations": ["min"],
# "case_uuid": TESTCASE_UUID,
# "class": "surface",
# "iteration_name": "iter-0",
# }
# with pytest.raises(Exception, match="40*"):
# response = explorer._sumo.post(f"/aggregations", json=body)
# print("Execution should never reach this line")
# print("Unexpected status: ", response.status_code)
# print("Unexpected response: ", response.text)


def test_aggregations_fast(explorer: Explorer):
Expand Down
54 changes: 27 additions & 27 deletions tests/test_access/tst_access_drogon_manage_login.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,33 +96,33 @@ def test_read_restricted_classification_data(explorer: Explorer):
print("Hits on restricted:", hits)
assert hits > 0


@pytest.mark.skipif(not (sys.platform == "linux" and
sys.version_info[:2] == (3, 11)),
reason="Test only on single platform/version.")
def test_aggregations_bulk(explorer: Explorer):
"""Test a bulk aggregation method"""
print("Running test:", inspect.currentframe().f_code.co_name)
cases = explorer.cases.filter(status="scratch")
print("Number of cases: ", len(cases))
assert len(cases) > 0
case = None
for c in cases:
if len(c.realizations) > 1 and len(c.surfaces) > 40:
case = c
break
assert case
case_uuid = case.metadata.get("fmu").get("case").get("uuid")
print("About to trigger aggregation on case", case_uuid)
body = {
"operations": ["min"],
"case_uuid": case_uuid,
"class": "surface",
"iteration_name": case.iterations[0].name,
}
response = explorer._sumo.post(f"/aggregations", json=body)
print(response.status_code)
assert response.status_code in [200, 201, 202]
# Remove or update this test when bulk aggregation is finalized
# @pytest.mark.skipif(not (sys.platform == "linux" and
# sys.version_info[:2] == (3, 11)),
# reason="Test only on single platform/version.")
# def test_aggregations_bulk(explorer: Explorer):
# """Test a bulk aggregation method"""
# print("Running test:", inspect.currentframe().f_code.co_name)
# cases = explorer.cases.filter(status="scratch")
# print("Number of cases: ", len(cases))
# assert len(cases) > 0
# case = None
# for c in cases:
# if len(c.realizations) > 1 and len(c.surfaces) > 40:
# case = c
# break
# assert case
# case_uuid = case.metadata.get("fmu").get("case").get("uuid")
# print("About to trigger aggregation on case", case_uuid)
# body = {
# "operations": ["min"],
# "case_uuid": case_uuid,
# "class": "surface",
# "iteration_name": case.iterations[0].name,
# }
# response = explorer._sumo.post(f"/aggregations", json=body)
# print(response.status_code)
# assert response.status_code in [200, 201, 202]


def test_aggregations_fast(explorer: Explorer):
Expand Down
62 changes: 31 additions & 31 deletions tests/test_access/tst_access_drogon_read_login.py
Original file line number Diff line number Diff line change
Expand Up @@ -150,37 +150,37 @@ def test_aggregations_fast(explorer: Explorer):
assert response.status_code in [200, 201, 202]
print("Length of returned aggregate object:", len(response.text))


@pytest.mark.skipif(not (sys.platform == "linux" and
sys.version_info[:2] == (3, 11)),
reason="Test only on single platform/version.")
def test_aggregate_bulk(explorer: Explorer):
"""Test a bulk aggregation method"""
print("Running test:", inspect.currentframe().f_code.co_name)
cases = explorer.cases
print("Number of cases: ", len(cases))
assert len(cases) > 0
case = None
for c in cases:
if len(c.realizations) > 1 and len(c.surfaces) > 40:
case = c
break
assert case
case_uuid = case.metadata.get("fmu").get("case").get("uuid")
body = {
"operations": ["min"],
"case_uuid": case_uuid,
"class": "surface",
"iteration_name": case.iterations[0].name,
}
print("About to trigger bulk aggregation on case", case_uuid)
print("using body", body, " this should raise exception")
# A READ role user CANNOT trigger BULK aggregation
with pytest.raises(Exception, match="403*"):
response = explorer._sumo.post(f"/aggregations", json=body)
print("Execution should never reach this line")
print("Unexpected status: ", response.status_code)
print("Unexpected response: ", response.text)
# Remove or update this test when bulk aggregation is finalized
# @pytest.mark.skipif(not (sys.platform == "linux" and
# sys.version_info[:2] == (3, 11)),
# reason="Test only on single platform/version.")
# def test_aggregate_bulk(explorer: Explorer):
# """Test a bulk aggregation method"""
# print("Running test:", inspect.currentframe().f_code.co_name)
# cases = explorer.cases
# print("Number of cases: ", len(cases))
# assert len(cases) > 0
# case = None
# for c in cases:
# if len(c.realizations) > 1 and len(c.surfaces) > 40:
# case = c
# break
# assert case
# case_uuid = case.metadata.get("fmu").get("case").get("uuid")
# body = {
# "operations": ["min"],
# "case_uuid": case_uuid,
# "class": "surface",
# "iteration_name": case.iterations[0].name,
# }
# print("About to trigger bulk aggregation on case", case_uuid)
# print("using body", body, " this should raise exception")
# # A READ role user CANNOT trigger BULK aggregation
# with pytest.raises(Exception, match="403*"):
# response = explorer._sumo.post(f"/aggregations", json=body)
# print("Execution should never reach this line")
# print("Unexpected status: ", response.status_code)
# print("Unexpected response: ", response.text)


def test_get_access_log(explorer: Explorer):
Expand Down
54 changes: 27 additions & 27 deletions tests/test_access/tst_access_drogon_write_login.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,33 +96,33 @@ def test_read_restricted_classification_data(explorer: Explorer):
print("Hits on restricted:", hits)
assert hits > 0


@pytest.mark.skipif(not (sys.platform == "linux" and
sys.version_info[:2] == (3, 11)),
reason="Test only on single platform/version.")
def test_aggregate_bulk(explorer: Explorer):
"""Test a bulk aggregation method"""
print("Running test:", inspect.currentframe().f_code.co_name)
cases = explorer.cases.filter(status="scratch")
print("Number of cases: ", len(cases))
assert len(cases) > 0
case = None
for c in cases:
if len(c.realizations) > 1 and len(c.surfaces) > 40:
case = c
break
assert case
case_uuid = case.metadata.get("fmu").get("case").get("uuid")
print("About to trigger aggregation on case", case_uuid)
body = {
"operations": ["min"],
"case_uuid": case_uuid,
"class": "surface",
"iteration_name": case.iterations[0].name,
}
response = explorer._sumo.post(f"/aggregations", json=body)
print(response.status_code)
assert response.status_code in [200, 201, 202]
# Remove or update this test when bulk aggregation is finalized
# @pytest.mark.skipif(not (sys.platform == "linux" and
# sys.version_info[:2] == (3, 11)),
# reason="Test only on single platform/version.")
# def test_aggregate_bulk(explorer: Explorer):
# """Test a bulk aggregation method"""
# print("Running test:", inspect.currentframe().f_code.co_name)
# cases = explorer.cases.filter(status="scratch")
# print("Number of cases: ", len(cases))
# assert len(cases) > 0
# case = None
# for c in cases:
# if len(c.realizations) > 1 and len(c.surfaces) > 40:
# case = c
# break
# assert case
# case_uuid = case.metadata.get("fmu").get("case").get("uuid")
# print("About to trigger aggregation on case", case_uuid)
# body = {
# "operations": ["min"],
# "case_uuid": case_uuid,
# "class": "surface",
# "iteration_name": case.iterations[0].name,
# }
# response = explorer._sumo.post(f"/aggregations", json=body)
# print(response.status_code)
# assert response.status_code in [200, 201, 202]


def test_aggregations_fast(explorer: Explorer):
Expand Down
42 changes: 21 additions & 21 deletions tests/test_access/tst_access_no_access_login.py
Original file line number Diff line number Diff line change
Expand Up @@ -157,27 +157,27 @@ def test_get_message_log_truncate(explorer: Explorer):
print("Unexpected status: ", response.status_code)
print("Unexpected response: ", response.text)


@pytest.mark.skipif(not (sys.platform == "linux" and
sys.version_info[:2] == (3, 11)),
reason="Test only on single platform/version.")
def test_aggregate_bulk(explorer: Explorer):
"""Test a bulk aggregation method"""
print("Running test:", inspect.currentframe().f_code.co_name)
# Fixed test case ("Drogon_AHM_2023-02-22") in Sumo/DEV
TESTCASE_UUID = "10f41041-2c17-4374-a735-bb0de62e29dc"
print("About to trigger bulk aggregation on case", TESTCASE_UUID)
body = {
"operations": ["min"],
"case_uuid": TESTCASE_UUID,
"class": "surface",
"iteration_name": "iter-0",
}
with pytest.raises(Exception, match="40*"):
response = explorer._sumo.post(f"/aggregations", json=body)
print("Execution should never reach this line")
print("Unexpected status: ", response.status_code)
print("Unexpected response: ", response.text)
# Remove or update this test when bulk aggregation is finalized
# @pytest.mark.skipif(not (sys.platform == "linux" and
# sys.version_info[:2] == (3, 11)),
# reason="Test only on single platform/version.")
# def test_aggregate_bulk(explorer: Explorer):
# """Test a bulk aggregation method"""
# print("Running test:", inspect.currentframe().f_code.co_name)
# # Fixed test case ("Drogon_AHM_2023-02-22") in Sumo/DEV
# TESTCASE_UUID = "10f41041-2c17-4374-a735-bb0de62e29dc"
# print("About to trigger bulk aggregation on case", TESTCASE_UUID)
# body = {
# "operations": ["min"],
# "case_uuid": TESTCASE_UUID,
# "class": "surface",
# "iteration_name": "iter-0",
# }
# with pytest.raises(Exception, match="40*"):
# response = explorer._sumo.post(f"/aggregations", json=body)
# print("Execution should never reach this line")
# print("Unexpected status: ", response.status_code)
# print("Unexpected response: ", response.text)


def test_aggregations_fast(explorer: Explorer):
Expand Down

0 comments on commit 126b6c5

Please sign in to comment.