Skip to content

DVPL-12514: Improve existing CSC app tests #633

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Aug 19, 2025
Merged

DVPL-12514: Improve existing CSC app tests #633

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ccf4e8e
fix csc apps test
Aug 8, 2025
2223825
simplify eventing app test
Aug 18, 2025
23f0263
split csc tests into behavior and metadata
Aug 18, 2025
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
119 changes: 76 additions & 43 deletions tests/searchcommands/test_csc_apps.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,16 +22,17 @@


@pytest.mark.smoke
class TestCSC(testlib.SDKTestCase):
def test_eventing_app(self):
app_name = "eventing_app"
class TestEventingApp(testlib.SDKTestCase):
app_name = "eventing_app"

def test_metadata(self):
self.assertTrue(
app_name in self.service.apps, msg="%s is not installed." % app_name
TestEventingApp.app_name in self.service.apps,
msg=f"{TestEventingApp.app_name} is not installed.",
)

# Fetch the app
app = self.service.apps[app_name]
app = self.service.apps[TestEventingApp.app_name]
app.refresh()

# Extract app info
Expand All @@ -40,6 +41,8 @@ def test_eventing_app(self):
state = app.state

# App info assertions
self.assertEqual(state.title, TestEventingApp.app_name)

self.assertEqual(access.app, "system")
self.assertEqual(access.can_change_perms, "1")
self.assertEqual(access.can_list, "1")
Expand All @@ -61,30 +64,51 @@ def test_eventing_app(self):
self.assertEqual(content.version, "1.0.0")
self.assertEqual(content.visible, "1")

self.assertEqual(state.title, "eventing_app")

jobs = self.service.jobs
stream = jobs.oneshot(
'search index="_internal" | head 4000 | eventingcsc status=200 | head 10',
output_mode="json",
def test_behavior(self):
makeresults_count = 20
expected_results_count = 10
expected_status = "200"

search_query = f"""
| makeresults count={makeresults_count}
| streamstats count as row_num
| eval status=case(
(row_num % 2) == 1, 200,
1=1, 500
)
result = results.JSONResultsReader(stream)
ds = list(result)
| eventingcsc status={expected_status}
"""
stream = self.service.jobs.oneshot(search_query, output_mode="json")

results_reader = results.JSONResultsReader(stream)
items = list(results_reader)

self.assertEqual(result.is_preview, False)
self.assertTrue(isinstance(ds[0], (dict, results.Message)))
nonmessages = [d for d in ds if isinstance(d, dict)]
self.assertTrue(len(nonmessages) <= 10)
self.assertFalse(results_reader.is_preview)

def test_generating_app(self):
app_name = "generating_app"
# filter out informational messages and keep only search results
actual_results = [
item for item in items if not isinstance(item, results.Message)
]

self.assertTrue(len(actual_results) == expected_results_count)

for res in actual_results:
self.assertIn("status", res)
self.assertEqual(res["status"], expected_status)


@pytest.mark.smoke
class TestGeneratingApp(testlib.SDKTestCase):
app_name = "generating_app"

def test_metadata(self):
self.assertTrue(
app_name in self.service.apps, msg="%s is not installed." % app_name
TestGeneratingApp.app_name in self.service.apps,
msg=f"{TestGeneratingApp.app_name} is not installed.",
)

# Fetch the app
app = self.service.apps[app_name]
app = self.service.apps[TestGeneratingApp.app_name]
app.refresh()

# Extract app info
Expand All @@ -93,6 +117,8 @@ def test_generating_app(self):
state = app.state

# App info assertions
self.assertEqual(state.title, TestGeneratingApp.app_name)

self.assertEqual(access.app, "system")
self.assertEqual(access.can_change_perms, "1")
self.assertEqual(access.can_list, "1")
Expand All @@ -116,23 +142,27 @@ def test_generating_app(self):
self.assertEqual(content.version, "1.0.0")
self.assertEqual(content.visible, "1")

self.assertEqual(state.title, "generating_app")

jobs = self.service.jobs
stream = jobs.oneshot("| generatingcsc count=4", output_mode="json")
def test_behavior(self):
stream = self.service.jobs.oneshot(
"| generatingcsc count=4", output_mode="json"
)
result = results.JSONResultsReader(stream)
ds = list(result)
self.assertTrue(len(ds) == 4)

def test_reporting_app(self):
app_name = "reporting_app"

@pytest.mark.smoke
class TestReportingApp(testlib.SDKTestCase):
app_name = "reporting_app"

def test_metadata(self):
self.assertTrue(
app_name in self.service.apps, msg="%s is not installed." % app_name
TestReportingApp.app_name in self.service.apps,
msg=f"{TestReportingApp.app_name} is not installed.",
)

# Fetch the app
app = self.service.apps[app_name]
app = self.service.apps[TestReportingApp.app_name]
app.refresh()

# Extract app info
Expand All @@ -141,6 +171,8 @@ def test_reporting_app(self):
state = app.state

# App info assertions
self.assertEqual(state.title, TestReportingApp.app_name)

self.assertEqual(access.app, "system")
self.assertEqual(access.can_change_perms, "1")
self.assertEqual(access.can_list, "1")
Expand All @@ -164,11 +196,9 @@ def test_reporting_app(self):
self.assertEqual(content.version, "1.0.0")
self.assertEqual(content.visible, "1")

self.assertEqual(state.title, "reporting_app")

def test_behavior_all_entries_above_cutoff(self):
jobs = self.service.jobs

# All above 150
stream = jobs.oneshot(
"| makeresults count=10 | eval math=100, eng=100, cs=100 | reportingcsc cutoff=150 math eng cs",
output_mode="json",
Expand All @@ -183,8 +213,8 @@ def test_reporting_app(self):
no_of_students = int(list(ds[0].values())[0])
self.assertTrue(no_of_students == 10)

# All below 150
stream = jobs.oneshot(
def test_behavior_all_entries_below_cutoff(self):
stream = self.service.jobs.oneshot(
"| makeresults count=10 | eval math=45, eng=45, cs=45 | reportingcsc cutoff=150 math eng cs",
output_mode="json",
)
Expand All @@ -198,15 +228,19 @@ def test_reporting_app(self):
no_of_students = int(list(ds[0].values())[0])
self.assertTrue(no_of_students == 0)

def test_streaming_app(self):
app_name = "streaming_app"

@pytest.mark.smoke
class TestStreamingApp(testlib.SDKTestCase):
app_name = "streaming_app"

def test_metadata(self):
self.assertTrue(
app_name in self.service.apps, msg="%s is not installed." % app_name
TestStreamingApp.app_name in self.service.apps,
msg=f"{TestStreamingApp.app_name} is not installed.",
)

# Fetch the app
app = self.service.apps[app_name]
app = self.service.apps[TestStreamingApp.app_name]
app.refresh()

# Extract app info
Expand All @@ -215,6 +249,8 @@ def test_streaming_app(self):
state = app.state

# App info assertions
self.assertEqual(state.title, TestStreamingApp.app_name)

self.assertEqual(access.app, "system")
self.assertEqual(access.can_change_perms, "1")
self.assertEqual(access.can_list, "1")
Expand All @@ -238,11 +274,8 @@ def test_streaming_app(self):
self.assertEqual(content.version, "1.0.0")
self.assertEqual(content.visible, "1")

self.assertEqual(state.title, "streaming_app")

jobs = self.service.jobs

stream = jobs.oneshot(
def test_behavior(self):
stream = self.service.jobs.oneshot(
"| makeresults count=5 | eval celsius = 35 | streamingcsc",
output_mode="json",
)
Expand Down