Skip to content

Commit 70a0cb3

Browse files
fix unit test of sender
1 parent d6d1b43 commit 70a0cb3

File tree

1 file changed

+78
-76
lines changed

1 file changed

+78
-76
lines changed

tests/test_radas_send_handler.py

Lines changed: 78 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -1,96 +1,98 @@
11
import tempfile
22
import os
33
from unittest import mock
4+
import unittest
45
from charon.pkgs.radas_signature_handler import sign_in_radas
56

67

7-
def test_sign_in_radas_normal_flow(self):
8-
with tempfile.TemporaryDirectory() as tmpdir:
9-
# Mock configuration
8+
class RadasSignHandlerTest(unittest.TestCase):
9+
def setUp(self) -> None:
10+
super().setUp()
11+
12+
def tearDown(self) -> None:
13+
super().tearDown()
14+
15+
def test_sign_in_radas_normal_flow(self):
16+
with tempfile.TemporaryDirectory() as tmpdir:
17+
# Mock configuration
18+
mock_config = mock.MagicMock()
19+
mock_config.is_radas_enabled.return_value = True
20+
mock_radas_config = mock.MagicMock()
21+
mock_config.get_radas_config.return_value = mock_radas_config
22+
23+
# Mock Container run to avoid real AMQP connection
24+
with mock.patch(
25+
"charon.pkgs.radas_signature_handler.Container") as mock_container, \
26+
mock.patch(
27+
"charon.pkgs.radas_signature_handler.get_config", return_value=mock_config), \
28+
mock.patch(
29+
"charon.pkgs.radas_signature_handler.uuid.uuid4", return_value="mocked-uuid"):
30+
31+
test_result_path = os.path.join(tmpdir, "results")
32+
os.makedirs(test_result_path)
33+
34+
sign_in_radas(
35+
repo_url="quay.io/test/repo",
36+
requester="test-user",
37+
sign_key="test-key",
38+
result_path=test_result_path,
39+
ignore_patterns=[],
40+
radas_config=mock_radas_config
41+
)
42+
43+
# Verify Container.run() was called twice (sender and receiver)
44+
self.assertEqual(mock_container.call_count, 2)
45+
46+
# Verify request ID propagation
47+
receiver_call = mock_container.call_args_list[1]
48+
self.assertEqual(receiver_call.args[0].request_id, "mocked-uuid")
49+
50+
51+
def test_sign_in_radas_with_disabled_config(self):
52+
mock_config = mock.MagicMock()
53+
mock_config.is_radas_enabled.return_value = False
54+
55+
with mock.patch("charon.pkgs.radas_signature_handler.get_config", return_value=mock_config), \
56+
self.assertRaises(SystemExit):
57+
sign_in_radas(
58+
repo_url="quay.io/test/repo",
59+
requester="test-user",
60+
sign_key="test-key",
61+
result_path="/tmp/results",
62+
ignore_patterns=[],
63+
radas_config=mock.MagicMock()
64+
)
65+
66+
67+
def test_sign_in_radas_connection_cleanup(self):
1068
mock_config = mock.MagicMock()
1169
mock_config.is_radas_enabled.return_value = True
1270
mock_radas_config = mock.MagicMock()
13-
mock_config.get_radas_config.return_value = mock_radas_config
1471

15-
# Mock Container run to avoid real AMQP connection
16-
with mock.patch(
17-
"charon.pkgs.radas_signature_handler.Container") as mock_container, \
18-
mock.patch(
19-
"charon.pkgs.radas_signature_handler.get_config", return_value=mock_config), \
20-
mock.patch(
21-
"charon.pkgs.radas_signature_handler.uuid.uuid4", return_value="mocked-uuid"):
72+
with mock.patch("charon.pkgs.radas_signature_handler.Container") as mock_container, \
73+
mock.patch("charon.pkgs.radas_signature_handler.get_config", return_value=mock_config):
74+
75+
mock_sender_conn = mock.MagicMock()
76+
mock_listener_conn = mock.MagicMock()
77+
78+
def container_side_effect(*args, **kwargs):
79+
if args[0].__class__.__name__ == "RadasReceiver":
80+
args[0].conn = mock_listener_conn
81+
elif args[0].__class__.__name__ == "RadasSender":
82+
args[0].conn = mock_sender_conn
83+
return mock.MagicMock()
2284

23-
# Test parameters
24-
test_result_path = os.path.join(tmpdir, "results")
25-
os.makedirs(test_result_path)
85+
mock_container.side_effect = container_side_effect
2686

27-
# Call the function
2887
sign_in_radas(
2988
repo_url="quay.io/test/repo",
3089
requester="test-user",
3190
sign_key="test-key",
32-
result_path=test_result_path,
91+
result_path="/tmp/results",
3392
ignore_patterns=[],
3493
radas_config=mock_radas_config
3594
)
3695

37-
# Verify Container.run() was called twice (sender and receiver)
38-
self.assertEqual(mock_container.call_count, 2)
39-
40-
# Verify request ID propagation
41-
receiver_call = mock_container.call_args_list[1]
42-
self.assertEqual(receiver_call.args[0].request_id, "mocked-uuid")
43-
44-
45-
def test_sign_in_radas_with_disabled_config(self):
46-
# Mock disabled configuration
47-
mock_config = mock.MagicMock()
48-
mock_config.is_radas_enabled.return_value = False
49-
50-
with mock.patch("charon.pkgs.radas_signature_handler.get_config", return_value=mock_config), \
51-
self.assertRaises(SystemExit):
52-
sign_in_radas(
53-
repo_url="quay.io/test/repo",
54-
requester="test-user",
55-
sign_key="test-key",
56-
result_path="/tmp/results",
57-
ignore_patterns=[],
58-
radas_config=mock.MagicMock()
59-
)
60-
61-
62-
def test_sign_in_radas_connection_cleanup(self):
63-
# Mock configuration and connection objects
64-
mock_config = mock.MagicMock()
65-
mock_config.is_radas_enabled.return_value = True
66-
mock_radas_config = mock.MagicMock()
67-
68-
with mock.patch("charon.pkgs.radas_signature_handler.Container") as mock_container, \
69-
mock.patch("charon.pkgs.radas_signature_handler.get_config", return_value=mock_config):
70-
71-
# Mock connection objects
72-
mock_sender_conn = mock.MagicMock()
73-
mock_listener_conn = mock.MagicMock()
74-
75-
# Create test call
76-
def container_side_effect(*args, **kwargs):
77-
if args[0].__class__.__name__ == "RadasReceiver":
78-
args[0].conn = mock_listener_conn
79-
elif args[0].__class__.__name__ == "RadasSender":
80-
args[0].conn = mock_sender_conn
81-
return mock.MagicMock()
82-
83-
mock_container.side_effect = container_side_effect
84-
85-
sign_in_radas(
86-
repo_url="quay.io/test/repo",
87-
requester="test-user",
88-
sign_key="test-key",
89-
result_path="/tmp/results",
90-
ignore_patterns=[],
91-
radas_config=mock_radas_config
92-
)
93-
94-
# Verify connections are closed
95-
mock_sender_conn.close.assert_called_once()
96-
mock_listener_conn.close.assert_called_once()
96+
# Verify connections are closed
97+
mock_sender_conn.close.assert_called_once()
98+
mock_listener_conn.close.assert_called_once()

0 commit comments

Comments
 (0)