3227 lines
117 KiB
Python
3227 lines
117 KiB
Python
# Copyright (C) 2024 Cetmix OÜ
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
|
|
|
|
from odoo.exceptions import ValidationError
|
|
|
|
from .common_jets import TestTowerJetsCommon
|
|
|
|
|
|
class TestTowerJetTemplate(TestTowerJetsCommon):
|
|
"""
|
|
Test the jet template model
|
|
"""
|
|
|
|
# All jet-related test data is now inherited from TestTowerJetsCommon
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
|
|
# Create additional servers for multi-server tests
|
|
cls.server_test_2 = cls.Server.create(
|
|
{
|
|
"name": "Test Server 2",
|
|
"reference": "test_server_2",
|
|
"ip_v4_address": "192.168.1.102",
|
|
"ssh_username": "admin",
|
|
"ssh_password": "password",
|
|
"ssh_auth_mode": "p",
|
|
"os_id": cls.os_debian_10.id,
|
|
}
|
|
)
|
|
cls.server_test_3 = cls.Server.create(
|
|
{
|
|
"name": "Test Server 3",
|
|
"reference": "test_server_3",
|
|
"ip_v4_address": "192.168.1.103",
|
|
"ssh_username": "admin",
|
|
"ssh_password": "password",
|
|
"ssh_auth_mode": "p",
|
|
"os_id": cls.os_debian_10.id,
|
|
}
|
|
)
|
|
|
|
def test_compute_border_actions_no_actions(self):
|
|
"""
|
|
Test _compute_border_actions with no actions defined
|
|
"""
|
|
# Create a jet template with no actions
|
|
template = self.JetTemplate.create(
|
|
{
|
|
"name": "No Actions Template",
|
|
"reference": "no_actions_template",
|
|
"server_ids": [(4, self.server_test_1.id)],
|
|
}
|
|
)
|
|
|
|
# Both border actions should be False
|
|
self.assertFalse(
|
|
template.action_create_id,
|
|
"Create action should be False when no actions exist",
|
|
)
|
|
self.assertFalse(
|
|
template.action_destroy_id,
|
|
"Destroy action should be False when no actions exist",
|
|
)
|
|
|
|
def test_compute_border_actions_both_valid_actions(self):
|
|
"""
|
|
Test _compute_border_actions with both valid create and destroy actions
|
|
"""
|
|
# Use common actions from class setup
|
|
create_action = self.action_create
|
|
destroy_action = self.action_destroy
|
|
|
|
# Both actions should be set
|
|
self.assertEqual(
|
|
self.jet_template_test.action_create_id,
|
|
create_action,
|
|
"Create action should be set to the valid action",
|
|
)
|
|
self.assertEqual(
|
|
self.jet_template_test.action_destroy_id,
|
|
destroy_action,
|
|
"Destroy action should be set to the valid action",
|
|
)
|
|
|
|
def test_compute_border_actions_invalid_create_action_with_initial_state(self):
|
|
"""
|
|
Test _compute_border_actions with invalid create action (has initial state)
|
|
"""
|
|
# Create an invalid create action (has state_from_id)
|
|
invalid_create_action = self.JetAction.create(
|
|
{
|
|
"name": "Invalid Create Action",
|
|
"reference": "invalid_create_action",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_initial.id, # Invalid for create
|
|
"state_to_id": self.state_running.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Since action_create_id is readonly=False, we can set it directly
|
|
# but the compute method won't be triggered automatically
|
|
self.jet_template_test.action_create_id = invalid_create_action
|
|
|
|
# The action should remain set because compute method wasn't triggered
|
|
self.assertEqual(
|
|
self.jet_template_test.action_create_id,
|
|
invalid_create_action,
|
|
"Create action should remain set when directly assigned (readonly=False)",
|
|
)
|
|
|
|
# Now trigger the compute method manually to test the logic
|
|
self.jet_template_test._compute_border_actions()
|
|
|
|
# Create action should be cleared because it's invalid
|
|
self.assertFalse(
|
|
self.jet_template_test.action_create_id,
|
|
"Create action should be cleared when it has an initial state",
|
|
)
|
|
|
|
def test_compute_border_actions_invalid_create_action_no_final_state(self):
|
|
"""
|
|
Test _compute_border_actions with invalid create action (no final state)
|
|
"""
|
|
# Create an invalid create action (no state_to_id)
|
|
invalid_create_action = self.JetAction.create(
|
|
{
|
|
"name": "Invalid Create Action",
|
|
"reference": "invalid_create_action",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": False,
|
|
"state_to_id": False, # No final state - invalid for create
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Since action_create_id is readonly=False, we can set it directly
|
|
# but the compute method won't be triggered automatically
|
|
self.jet_template_test.action_create_id = invalid_create_action
|
|
|
|
# The action should remain set because compute method wasn't triggered
|
|
self.assertEqual(
|
|
self.jet_template_test.action_create_id,
|
|
invalid_create_action,
|
|
"Create action should remain set when directly assigned (readonly=False)",
|
|
)
|
|
|
|
# Now trigger the compute method manually to test the logic
|
|
self.jet_template_test._compute_border_actions()
|
|
|
|
# Create action should be cleared because it's invalid
|
|
self.assertFalse(
|
|
self.jet_template_test.action_create_id,
|
|
"Create action should be cleared when it has no final state",
|
|
)
|
|
|
|
def test_compute_border_actions_invalid_destroy_action_with_final_state(self):
|
|
"""
|
|
Test _compute_border_actions with invalid destroy action (has final state)
|
|
"""
|
|
# Create an invalid destroy action (has state_to_id)
|
|
invalid_destroy_action = self.JetAction.create(
|
|
{
|
|
"name": "Invalid Destroy Action",
|
|
"reference": "invalid_destroy_action",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_running.id,
|
|
"state_to_id": self.state_stopped.id, # Invalid for destroy
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Since action_destroy_id is readonly=False, we can set it directly
|
|
# but the compute method won't be triggered automatically
|
|
self.jet_template_test.action_destroy_id = invalid_destroy_action
|
|
|
|
# The action should remain set because compute method wasn't triggered
|
|
self.assertEqual(
|
|
self.jet_template_test.action_destroy_id,
|
|
invalid_destroy_action,
|
|
"Destroy action should remain set when directly assigned (readonly=False)",
|
|
)
|
|
|
|
# Now trigger the compute method manually to test the logic
|
|
self.jet_template_test._compute_border_actions()
|
|
|
|
# Destroy action should be cleared because it's invalid
|
|
self.assertFalse(
|
|
self.jet_template_test.action_destroy_id,
|
|
"Destroy action should be cleared when it has a final state",
|
|
)
|
|
|
|
def test_compute_border_actions_multiple_actions_priority(self):
|
|
"""
|
|
Test _compute_border_actions with multiple actions, checking priority order
|
|
"""
|
|
# Clear existing border actions to force recomputation
|
|
self.jet_template_test.action_create_id = False
|
|
self.jet_template_test.action_destroy_id = False
|
|
|
|
# Create multiple create actions with different priorities
|
|
# Use priority 0 to ensure they have higher priority
|
|
# than common actions (priority 1)
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Create Action 1",
|
|
"reference": "create_action_1",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": False,
|
|
"state_to_id": self.state_running.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 2, # Higher priority number (lower priority)
|
|
}
|
|
)
|
|
|
|
create_action_2 = self.JetAction.create(
|
|
{
|
|
"name": "Create Action 2",
|
|
"reference": "create_action_2",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": False,
|
|
"state_to_id": self.state_running.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 0, # Lower priority number (higher priority)
|
|
}
|
|
)
|
|
|
|
# Create multiple destroy actions with different priorities
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Destroy Action 1",
|
|
"reference": "destroy_action_1",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_running.id,
|
|
"state_to_id": False,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 2, # Higher priority number (lower priority)
|
|
}
|
|
)
|
|
|
|
destroy_action_2 = self.JetAction.create(
|
|
{
|
|
"name": "Destroy Action 2",
|
|
"reference": "destroy_action_2",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_running.id,
|
|
"state_to_id": False,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 0, # Lower priority number (higher priority)
|
|
}
|
|
)
|
|
|
|
# Trigger recomputation of border actions to ensure
|
|
# the new actions are considered
|
|
self.jet_template_test._compute_border_actions()
|
|
|
|
# Should select the actions with higher priority (lower priority number)
|
|
self.assertEqual(
|
|
self.jet_template_test.action_create_id,
|
|
create_action_2,
|
|
"Create action should be the one with higher priority",
|
|
)
|
|
self.assertEqual(
|
|
self.jet_template_test.action_destroy_id,
|
|
destroy_action_2,
|
|
"Destroy action should be the one with higher priority",
|
|
)
|
|
|
|
def test_compute_border_actions_action_updates(self):
|
|
"""
|
|
Test _compute_border_actions when actions are updated
|
|
"""
|
|
# Use common actions from class setup
|
|
create_action = self.action_create
|
|
destroy_action = self.action_destroy
|
|
|
|
# Both actions should be set initially
|
|
self.assertEqual(self.jet_template_test.action_create_id, create_action)
|
|
self.assertEqual(self.jet_template_test.action_destroy_id, destroy_action)
|
|
|
|
# Update create action to make it invalid (add initial state)
|
|
create_action.write({"state_from_id": self.state_initial.id})
|
|
|
|
# Create action should be cleared, destroy action should remain
|
|
self.assertFalse(
|
|
self.jet_template_test.action_create_id,
|
|
"Create action should be cleared after becoming invalid",
|
|
)
|
|
self.assertEqual(
|
|
self.jet_template_test.action_destroy_id,
|
|
destroy_action,
|
|
"Destroy action should remain unchanged",
|
|
)
|
|
|
|
# Update destroy action to make it invalid (add final state)
|
|
destroy_action.write({"state_to_id": self.state_stopped.id})
|
|
|
|
# Both actions should be cleared
|
|
self.assertFalse(
|
|
self.jet_template_test.action_create_id,
|
|
"Create action should remain cleared",
|
|
)
|
|
self.assertFalse(
|
|
self.jet_template_test.action_destroy_id,
|
|
"Destroy action should be cleared after becoming invalid",
|
|
)
|
|
|
|
def test_find_action_path_bfs_multiple_paths_shortest(self):
|
|
"""
|
|
Test _find_action_path_bfs finds the shortest path when multiple paths exist
|
|
"""
|
|
# Create actions for multiple paths
|
|
# Short path: A -> C
|
|
action_ac = self.JetAction.create(
|
|
{
|
|
"name": "Action A to C (short)",
|
|
"reference": "action_ac",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
# Long path: A -> B -> D -> C
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_bd = self.JetAction.create(
|
|
{
|
|
"name": "Action B to D",
|
|
"reference": "action_bd",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_d.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_dc = self.JetAction.create(
|
|
{
|
|
"name": "Action D to C",
|
|
"reference": "action_dc",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_d.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Create adjacency with multiple paths
|
|
adjacency = {
|
|
self.state_a: [
|
|
(self.state_c, action_ac),
|
|
(self.state_b, action_ab),
|
|
], # Short and long path
|
|
self.state_b: [(self.state_d, action_bd)],
|
|
self.state_d: [(self.state_c, action_dc)],
|
|
}
|
|
|
|
# Test that shortest path is found
|
|
result = self.jet_template_test._find_action_path_bfs(
|
|
self.state_a, self.state_c, adjacency
|
|
)
|
|
expected_path = [action_ac] # Shortest path
|
|
self.assertEqual(
|
|
result,
|
|
expected_path,
|
|
"Should return shortest path when multiple paths exist",
|
|
)
|
|
|
|
def test_find_action_path_bfs_empty_adjacency(self):
|
|
"""
|
|
Test _find_action_path_bfs with empty adjacency list
|
|
"""
|
|
# Empty adjacency
|
|
adjacency = {}
|
|
|
|
# Test with empty adjacency
|
|
result = self.jet_template_test._find_action_path_bfs(
|
|
self.state_a, self.state_b, adjacency
|
|
)
|
|
self.assertIsNone(result, "Should return None with empty adjacency")
|
|
|
|
def test_find_action_path_bfs_cyclic_graph(self):
|
|
"""
|
|
Test _find_action_path_bfs with cyclic graph
|
|
"""
|
|
# Create actions for cyclic graph
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_bc = self.JetAction.create(
|
|
{
|
|
"name": "Action B to C",
|
|
"reference": "action_bc",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_ca = self.JetAction.create(
|
|
{
|
|
"name": "Action C to A",
|
|
"reference": "action_ca",
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"state_from_id": self.state_c.id,
|
|
"state_to_id": self.state_a.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Create cyclic adjacency: A -> B -> C -> A
|
|
adjacency = {
|
|
self.state_a: [(self.state_b, action_ab)],
|
|
self.state_b: [(self.state_c, action_bc)],
|
|
self.state_c: [(self.state_a, action_ca)],
|
|
}
|
|
|
|
# Test path from A to C (should find path despite cycle)
|
|
result = self.jet_template_test._find_action_path_bfs(
|
|
self.state_a, self.state_c, adjacency
|
|
)
|
|
expected_path = [action_ab, action_bc]
|
|
self.assertEqual(result, expected_path, "Should find path in cyclic graph")
|
|
|
|
def test_find_action_path_bfs_disconnected_states(self):
|
|
"""
|
|
Test _find_action_path_bfs with disconnected states
|
|
"""
|
|
# Create adjacency with disconnected components
|
|
adjacency = {
|
|
self.state_a: [(self.state_b, "action_ab")], # A and B connected
|
|
# state_c is isolated
|
|
}
|
|
|
|
# Test path from A to C (disconnected)
|
|
result = self.jet_template_test._find_action_path_bfs(
|
|
self.state_a, self.state_c, adjacency
|
|
)
|
|
self.assertIsNone(result, "Should return None for disconnected states")
|
|
|
|
def test_find_action_path_bfs_with_get_action_adjacency(self):
|
|
"""
|
|
Test _find_action_path_bfs using the actual _get_action_adjacency method
|
|
"""
|
|
# Create actions that will be used by _get_action_adjacency
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_bc = self.JetAction.create(
|
|
{
|
|
"name": "Action B to C",
|
|
"reference": "action_bc",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Get adjacency using the actual method
|
|
adjacency = self.clean_template._get_action_adjacency()
|
|
|
|
# Test path from A to C
|
|
result = self.clean_template._find_action_path_bfs(
|
|
self.state_a, self.state_c, adjacency
|
|
)
|
|
expected_path = [action_ab, action_bc]
|
|
self.assertEqual(
|
|
result, expected_path, "Should work with _get_action_adjacency method"
|
|
)
|
|
|
|
def test_get_action_adjacency_no_actions(self):
|
|
"""
|
|
Test _get_action_adjacency with no actions
|
|
"""
|
|
# Create a template with no actions
|
|
template = self.JetTemplate.create(
|
|
{
|
|
"name": "No Actions Template",
|
|
"reference": "no_actions_template",
|
|
"server_ids": [(4, self.server_test_1.id)],
|
|
}
|
|
)
|
|
|
|
# Get adjacency
|
|
adjacency = template._get_action_adjacency()
|
|
|
|
# Should return empty dict
|
|
self.assertEqual(
|
|
adjacency, {}, "Should return empty dict when no actions exist"
|
|
)
|
|
|
|
def test_get_action_adjacency_single_action(self):
|
|
"""
|
|
Test _get_action_adjacency with a single valid action
|
|
"""
|
|
# Create action
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Get adjacency
|
|
adjacency = self.clean_template._get_action_adjacency()
|
|
|
|
# Should have one entry
|
|
self.assertIn(self.state_a, adjacency, "Should include state_a in adjacency")
|
|
self.assertEqual(
|
|
len(adjacency[self.state_a]), 1, "Should have one transition from state_a"
|
|
)
|
|
self.assertEqual(
|
|
adjacency[self.state_a][0],
|
|
(self.state_b, action_ab),
|
|
"Should map to state_b with action_ab",
|
|
)
|
|
|
|
def test_get_action_adjacency_multiple_actions_from_same_state(self):
|
|
"""
|
|
Test _get_action_adjacency with multiple actions from the same state
|
|
"""
|
|
# Create multiple actions from state_a
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_ac = self.JetAction.create(
|
|
{
|
|
"name": "Action A to C",
|
|
"reference": "action_ac",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 20,
|
|
}
|
|
)
|
|
|
|
# Get adjacency
|
|
adjacency = self.clean_template._get_action_adjacency()
|
|
|
|
# Should have multiple transitions from state_a
|
|
self.assertIn(self.state_a, adjacency, "Should include state_a in adjacency")
|
|
self.assertEqual(
|
|
len(adjacency[self.state_a]), 2, "Should have two transitions from state_a"
|
|
)
|
|
|
|
# Check that both transitions are present
|
|
transitions = adjacency[self.state_a]
|
|
expected_transitions = [(self.state_b, action_ab), (self.state_c, action_ac)]
|
|
for expected in expected_transitions:
|
|
self.assertIn(
|
|
expected, transitions, f"Should include transition {expected}"
|
|
)
|
|
|
|
def test_get_action_adjacency_actions_without_from_state(self):
|
|
"""
|
|
Test _get_action_adjacency with actions that have no state_from_id
|
|
"""
|
|
# Create action without state_from_id (create action)
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Create Action",
|
|
"reference": "create_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": False, # No initial state
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Get adjacency
|
|
adjacency = self.clean_template._get_action_adjacency()
|
|
|
|
# Should be empty because action has no state_from_id
|
|
self.assertEqual(
|
|
adjacency, {}, "Should return empty dict for actions without state_from_id"
|
|
)
|
|
|
|
def test_get_action_adjacency_actions_without_to_state(self):
|
|
"""
|
|
Test _get_action_adjacency with actions that have no state_to_id
|
|
"""
|
|
# Create action without state_to_id (destroy action)
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Destroy Action",
|
|
"reference": "destroy_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": False, # No final state
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Get adjacency
|
|
adjacency = self.clean_template._get_action_adjacency()
|
|
|
|
# Should be empty because action has no state_to_id
|
|
self.assertEqual(
|
|
adjacency, {}, "Should return empty dict for actions without state_to_id"
|
|
)
|
|
|
|
def test_get_action_adjacency_complex_graph(self):
|
|
"""
|
|
Test _get_action_adjacency with a complex graph structure
|
|
"""
|
|
# Create complex action graph
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_ac = self.JetAction.create(
|
|
{
|
|
"name": "Action A to C",
|
|
"reference": "action_ac",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 20,
|
|
}
|
|
)
|
|
action_bd = self.JetAction.create(
|
|
{
|
|
"name": "Action B to D",
|
|
"reference": "action_bd",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_d.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_cd = self.JetAction.create(
|
|
{
|
|
"name": "Action C to D",
|
|
"reference": "action_cd",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_c.id,
|
|
"state_to_id": self.state_d.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Get adjacency
|
|
adjacency = self.clean_template._get_action_adjacency()
|
|
|
|
# Check structure
|
|
self.assertIn(self.state_a, adjacency, "Should include state_a")
|
|
self.assertIn(self.state_b, adjacency, "Should include state_b")
|
|
self.assertIn(self.state_c, adjacency, "Should include state_c")
|
|
self.assertNotIn(
|
|
self.state_d, adjacency, "Should not include state_d (no outgoing edges)"
|
|
)
|
|
|
|
# Check transitions from state_a
|
|
self.assertEqual(
|
|
len(adjacency[self.state_a]),
|
|
2,
|
|
"State A should have 2 outgoing transitions",
|
|
)
|
|
expected_from_a = [(self.state_b, action_ab), (self.state_c, action_ac)]
|
|
for expected in expected_from_a:
|
|
self.assertIn(
|
|
expected,
|
|
adjacency[self.state_a],
|
|
f"State A should have transition {expected}",
|
|
)
|
|
|
|
# Check transitions from state_b
|
|
self.assertEqual(
|
|
len(adjacency[self.state_b]), 1, "State B should have 1 outgoing transition"
|
|
)
|
|
self.assertEqual(
|
|
adjacency[self.state_b][0],
|
|
(self.state_d, action_bd),
|
|
"State B should transition to D",
|
|
)
|
|
|
|
# Check transitions from state_c
|
|
self.assertEqual(
|
|
len(adjacency[self.state_c]), 1, "State C should have 1 outgoing transition"
|
|
)
|
|
self.assertEqual(
|
|
adjacency[self.state_c][0],
|
|
(self.state_d, action_cd),
|
|
"State C should transition to D",
|
|
)
|
|
|
|
def test_get_action_adjacency_mixed_valid_invalid_actions(self):
|
|
"""
|
|
Test _get_action_adjacency with mix of valid and invalid actions
|
|
"""
|
|
# Create valid action
|
|
valid_action = self.JetAction.create(
|
|
{
|
|
"name": "Valid Action",
|
|
"reference": "valid_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Create invalid actions (should be ignored)
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Invalid Action 1",
|
|
"reference": "invalid_action_1",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": False, # No initial state
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Invalid Action 2",
|
|
"reference": "invalid_action_2",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": False, # No final state
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Get adjacency
|
|
adjacency = self.clean_template._get_action_adjacency()
|
|
|
|
# Should only include the valid action
|
|
self.assertIn(self.state_a, adjacency, "Should include state_a")
|
|
self.assertEqual(
|
|
len(adjacency[self.state_a]), 1, "Should have only one valid transition"
|
|
)
|
|
self.assertEqual(
|
|
adjacency[self.state_a][0],
|
|
(self.state_b, valid_action),
|
|
"Should include only valid action",
|
|
)
|
|
|
|
def test_get_action_adjacency_self_loop(self):
|
|
"""
|
|
Test _get_action_adjacency with self-loop actions
|
|
"""
|
|
# Create self-loop action
|
|
self_loop_action = self.JetAction.create(
|
|
{
|
|
"name": "Self Loop Action",
|
|
"reference": "self_loop_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_a.id, # Same state
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Get adjacency
|
|
adjacency = self.clean_template._get_action_adjacency()
|
|
|
|
# Should include self-loop
|
|
self.assertIn(self.state_a, adjacency, "Should include state_a")
|
|
self.assertEqual(
|
|
len(adjacency[self.state_a]), 1, "Should have one self-loop transition"
|
|
)
|
|
self.assertEqual(
|
|
adjacency[self.state_a][0],
|
|
(self.state_a, self_loop_action),
|
|
"Should include self-loop action",
|
|
)
|
|
|
|
def test_get_action_path_no_create_destroy_actions(self):
|
|
"""
|
|
Test _get_action_path when no create or destroy actions are set
|
|
"""
|
|
# Create a template with no border actions
|
|
template = self.JetTemplate.create(
|
|
{
|
|
"name": "No Border Actions Template",
|
|
"reference": "no_border_actions_template",
|
|
"server_ids": [(4, self.server_test_1.id)],
|
|
}
|
|
)
|
|
|
|
# Test path without state_from and state_to
|
|
result = template._get_action_path()
|
|
self.assertEqual(
|
|
result, [], "Should return empty list when no create action exists"
|
|
)
|
|
|
|
# Test path with state_from but no state_to
|
|
result = template._get_action_path(state_from=self.state_a)
|
|
self.assertEqual(
|
|
result, [], "Should return empty list when no destroy action exists"
|
|
)
|
|
|
|
def test_get_action_path_both_parameters_provided(self):
|
|
"""
|
|
Test _get_action_path when both state_from and state_to are provided
|
|
"""
|
|
# Create action
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Test path with both parameters provided
|
|
result = self.clean_template._get_action_path(
|
|
state_from=self.state_a, state_to=self.state_b
|
|
)
|
|
self.assertEqual(
|
|
result,
|
|
[action_ab],
|
|
"Should return action path when both parameters provided",
|
|
)
|
|
|
|
def test_get_action_path_requires_at_least_one_parameter(self):
|
|
"""
|
|
Test _get_action_path behavior when no parameters are provided
|
|
"""
|
|
# Create a template with no border actions
|
|
template = self.JetTemplate.create(
|
|
{
|
|
"name": "No Border Actions Template",
|
|
"reference": "no_border_actions_template",
|
|
"server_ids": [(4, self.server_test_1.id)],
|
|
}
|
|
)
|
|
|
|
# Test with no parameters - should return empty list
|
|
result = template._get_action_path()
|
|
self.assertEqual(
|
|
result,
|
|
[],
|
|
"Should return empty list when no parameters and no border actions",
|
|
)
|
|
|
|
# Test with only state_from
|
|
result = template._get_action_path(state_from=self.state_a)
|
|
self.assertEqual(
|
|
result,
|
|
[],
|
|
"Should return empty list when only state_from provided",
|
|
)
|
|
|
|
# Test with only state_to
|
|
result = template._get_action_path(state_to=self.state_b)
|
|
self.assertEqual(
|
|
result,
|
|
[],
|
|
"Should return empty list when only state_to provided and no create action",
|
|
)
|
|
|
|
def test_get_action_path_with_create_action_only(self):
|
|
"""
|
|
Test _get_action_path with only create action set
|
|
"""
|
|
# Create create action
|
|
create_action = self.JetAction.create(
|
|
{
|
|
"name": "Create Action",
|
|
"reference": "create_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": False,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Set create action
|
|
self.clean_template.action_create_id = create_action
|
|
|
|
# Test path without state_from (should return empty because no destroy action)
|
|
result = self.clean_template._get_action_path()
|
|
self.assertEqual(
|
|
result, [], "Should return empty list when no destroy action provided"
|
|
)
|
|
|
|
# Test path with state_from (should not use create action)
|
|
result = self.clean_template._get_action_path(state_from=self.state_b)
|
|
self.assertEqual(
|
|
result,
|
|
[],
|
|
"Should return empty list when state_from provided and no path exists",
|
|
)
|
|
|
|
def test_build_dependency_graph_simple_dependency(self):
|
|
"""Test _build_dependency_graph with simple dependency chain"""
|
|
# Use the existing dependency hierarchy
|
|
|
|
graph = self.jet_template_odoo._build_dependency_graph()
|
|
|
|
# Verify all templates are in the graph
|
|
expected_template_ids = [
|
|
self.jet_template_odoo.id,
|
|
self.jet_template_postgres.id,
|
|
self.jet_template_nginx.id,
|
|
self.jet_template_docker.id,
|
|
self.jet_template_tower_core.id,
|
|
]
|
|
self.assertEqual(
|
|
set(graph.keys()),
|
|
set(expected_template_ids),
|
|
"All templates should be in the graph",
|
|
)
|
|
|
|
# Verify Odoo template info
|
|
odoo_info = graph[self.jet_template_odoo.id]
|
|
self.assertEqual(odoo_info["template"], self.jet_template_odoo)
|
|
self.assertEqual(odoo_info["name"], "Odoo")
|
|
self.assertEqual(odoo_info["reference"], "odoo")
|
|
self.assertEqual(odoo_info["level"], 0) # Root template
|
|
self.assertEqual(
|
|
len(odoo_info["dependencies"]), 2
|
|
) # Depends on Postgres and Nginx
|
|
|
|
# Verify Odoo dependencies
|
|
odoo_dep_ids = [dep["template_id"] for dep in odoo_info["dependencies"]]
|
|
self.assertIn(self.jet_template_postgres.id, odoo_dep_ids)
|
|
self.assertIn(self.jet_template_nginx.id, odoo_dep_ids)
|
|
|
|
# Verify Postgres template info
|
|
postgres_info = graph[self.jet_template_postgres.id]
|
|
self.assertEqual(postgres_info["template"], self.jet_template_postgres)
|
|
self.assertEqual(postgres_info["name"], "Postgres")
|
|
self.assertEqual(postgres_info["reference"], "postgres")
|
|
self.assertEqual(postgres_info["level"], 1) # One level from root
|
|
self.assertEqual(len(postgres_info["dependencies"]), 1) # Depends on Docker
|
|
|
|
# Verify Postgres dependencies
|
|
postgres_dep_ids = [dep["template_id"] for dep in postgres_info["dependencies"]]
|
|
self.assertIn(self.jet_template_docker.id, postgres_dep_ids)
|
|
|
|
# Verify Nginx template info
|
|
nginx_info = graph[self.jet_template_nginx.id]
|
|
self.assertEqual(nginx_info["template"], self.jet_template_nginx)
|
|
self.assertEqual(nginx_info["name"], "Nginx")
|
|
self.assertEqual(nginx_info["reference"], "nginx")
|
|
self.assertEqual(nginx_info["level"], 1) # One level from root
|
|
self.assertEqual(len(nginx_info["dependencies"]), 1) # Depends on Docker
|
|
|
|
# Verify Nginx dependencies
|
|
nginx_dep_ids = [dep["template_id"] for dep in nginx_info["dependencies"]]
|
|
self.assertIn(self.jet_template_docker.id, nginx_dep_ids)
|
|
|
|
# Verify Docker template info
|
|
docker_info = graph[self.jet_template_docker.id]
|
|
self.assertEqual(docker_info["template"], self.jet_template_docker)
|
|
self.assertEqual(docker_info["name"], "Docker")
|
|
self.assertEqual(docker_info["reference"], "docker")
|
|
self.assertEqual(docker_info["level"], 2) # Two levels from root
|
|
self.assertEqual(len(docker_info["dependencies"]), 1) # Depends on Tower Core
|
|
|
|
# Verify Docker dependencies
|
|
docker_dep_ids = [dep["template_id"] for dep in docker_info["dependencies"]]
|
|
self.assertIn(self.jet_template_tower_core.id, docker_dep_ids)
|
|
|
|
# Verify Tower Core template info
|
|
tower_core_info = graph[self.jet_template_tower_core.id]
|
|
self.assertEqual(tower_core_info["template"], self.jet_template_tower_core)
|
|
self.assertEqual(tower_core_info["name"], "Tower Core")
|
|
self.assertEqual(tower_core_info["reference"], "tower_core")
|
|
self.assertEqual(tower_core_info["level"], 3) # Three levels from root
|
|
self.assertEqual(len(tower_core_info["dependencies"]), 0) # No dependencies
|
|
|
|
def test_build_dependency_graph_circular_dependency(self):
|
|
"""
|
|
Test circular dependency detection in constraint validation.
|
|
|
|
This test verifies that circular dependency detection correctly includes
|
|
the new dependency being created, not just existing ones from the database.
|
|
|
|
Scenario:
|
|
A->B, B->C exist, trying to create C->A should be detected as circular.
|
|
"""
|
|
|
|
# Create a circular dependency: A -> B -> C -> A
|
|
template_a = self.JetTemplate.create(
|
|
{
|
|
"name": "Template A",
|
|
"reference": "template_a",
|
|
}
|
|
)
|
|
template_b = self.JetTemplate.create(
|
|
{
|
|
"name": "Template B",
|
|
"reference": "template_b",
|
|
}
|
|
)
|
|
template_c = self.JetTemplate.create(
|
|
{
|
|
"name": "Template C",
|
|
"reference": "template_c",
|
|
}
|
|
)
|
|
|
|
# Create first two dependencies (A -> B -> C)
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": template_a.id,
|
|
"template_required_id": template_b.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": template_b.id,
|
|
"template_required_id": template_c.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
|
|
# The third dependency (C -> A) should raise a ValidationError
|
|
with self.assertRaises(ValidationError) as context:
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": template_c.id,
|
|
"template_required_id": template_a.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
|
|
# Verify the error message mentions circular reference
|
|
error_message = str(context.exception)
|
|
self.assertIn("circular reference", error_message.lower())
|
|
self.assertIn("Template C", error_message)
|
|
|
|
def test_build_dependency_graph_with_state_requirements(self):
|
|
"""Test _build_dependency_graph with state requirements"""
|
|
# pylint: disable=protected-access
|
|
# Create a template with state requirements
|
|
template_with_state = self.JetTemplate.create(
|
|
{
|
|
"name": "Template With State",
|
|
"reference": "template_with_state",
|
|
}
|
|
)
|
|
|
|
# Create dependency with state requirement
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": template_with_state.id,
|
|
"template_required_id": self.jet_template_tower_core.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
|
|
graph = template_with_state._build_dependency_graph()
|
|
|
|
# Verify the dependency includes state information
|
|
template_info = graph[template_with_state.id]
|
|
self.assertEqual(len(template_info["dependencies"]), 1)
|
|
|
|
dep_info = template_info["dependencies"][0]
|
|
self.assertEqual(dep_info["template_id"], self.jet_template_tower_core.id)
|
|
self.assertEqual(dep_info["template_name"], "Tower Core")
|
|
self.assertEqual(dep_info["template_reference"], "tower_core")
|
|
self.assertEqual(dep_info["required_state_id"], self.state_running.id)
|
|
self.assertEqual(dep_info["required_state_name"], "Test Running")
|
|
|
|
def test_build_dependency_graph_complex_hierarchy(self):
|
|
"""Test _build_dependency_graph with complex dependency hierarchy"""
|
|
# pylint: disable=protected-access
|
|
# Create a more complex hierarchy: E -> D, C -> B -> A
|
|
template_a = self.JetTemplate.create(
|
|
{
|
|
"name": "Template A",
|
|
"reference": "template_a",
|
|
}
|
|
)
|
|
template_b = self.JetTemplate.create(
|
|
{
|
|
"name": "Template B",
|
|
"reference": "template_b",
|
|
}
|
|
)
|
|
template_c = self.JetTemplate.create(
|
|
{
|
|
"name": "Template C",
|
|
"reference": "template_c",
|
|
}
|
|
)
|
|
template_d = self.JetTemplate.create(
|
|
{
|
|
"name": "Template D",
|
|
"reference": "template_d",
|
|
}
|
|
)
|
|
template_e = self.JetTemplate.create(
|
|
{
|
|
"name": "Template E",
|
|
"reference": "template_e",
|
|
}
|
|
)
|
|
|
|
# Create dependencies: E -> D, A -> B -> C
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": template_e.id,
|
|
"template_required_id": template_d.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": template_a.id,
|
|
"template_required_id": template_b.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": template_b.id,
|
|
"template_required_id": template_c.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
|
|
# Test from template E
|
|
graph = template_e._build_dependency_graph()
|
|
|
|
# Should contain E and D
|
|
expected_template_ids = [template_e.id, template_d.id]
|
|
self.assertEqual(
|
|
set(graph.keys()),
|
|
set(expected_template_ids),
|
|
"Should contain E and its dependencies",
|
|
)
|
|
|
|
# Verify levels
|
|
self.assertEqual(graph[template_e.id]["level"], 0) # Root
|
|
self.assertEqual(graph[template_d.id]["level"], 1) # One level down
|
|
|
|
# Test from template C
|
|
graph = template_c._build_dependency_graph()
|
|
|
|
# Should contain only C (C has no dependencies)
|
|
expected_template_ids = [template_c.id]
|
|
self.assertEqual(
|
|
set(graph.keys()), set(expected_template_ids), "Should contain only C"
|
|
)
|
|
|
|
# Verify levels
|
|
self.assertEqual(graph[template_c.id]["level"], 0) # Root
|
|
self.assertEqual(
|
|
len(graph[template_c.id]["dependencies"]), 0
|
|
) # No dependencies
|
|
|
|
# Test from template A - should include A, B, and C
|
|
# because A depends on B, and B depends on C
|
|
graph = template_a._build_dependency_graph()
|
|
|
|
# Should contain A, B, and C (A needs B, B needs C)
|
|
expected_template_ids = [template_a.id, template_b.id, template_c.id]
|
|
|
|
# Check that all expected templates are in the graph
|
|
for expected_id in expected_template_ids:
|
|
self.assertIn(
|
|
expected_id, graph, f"Template {expected_id} should be in the graph"
|
|
)
|
|
|
|
# Check that the graph contains at least the expected templates
|
|
# (it might contain more due to other templates in the test database)
|
|
self.assertTrue(
|
|
all(template_id in graph for template_id in expected_template_ids),
|
|
f"Graph should contain at least {expected_template_ids}",
|
|
)
|
|
|
|
# Verify levels for the expected templates
|
|
self.assertEqual(graph[template_a.id]["level"], 0) # Root
|
|
self.assertEqual(graph[template_b.id]["level"], 1) # One level down
|
|
self.assertEqual(graph[template_c.id]["level"], 2) # Two levels down
|
|
|
|
def test_build_dependency_graph_self_dependency(self):
|
|
"""Test _build_dependency_graph with self-dependency"""
|
|
|
|
# Create a template that depends on itself
|
|
template_self = self.JetTemplate.create(
|
|
{
|
|
"name": "Self Dependent Template",
|
|
"reference": "self_dependent_template",
|
|
}
|
|
)
|
|
|
|
# Creating self-dependency should raise a ValidationError
|
|
with self.assertRaises(ValidationError) as context:
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": template_self.id,
|
|
"template_required_id": template_self.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
|
|
# Verify the error message mentions self-dependency
|
|
error_message = str(context.exception)
|
|
self.assertIn("cannot depend on itself", error_message.lower())
|
|
|
|
def test_calculate_dependency_levels_simple_chain(self):
|
|
"""Test _calculate_dependency_levels with simple dependency chain"""
|
|
# pylint: disable=protected-access
|
|
# Use existing dependency chain: Odoo -> Postgres -> Docker -> Tower Core
|
|
|
|
# Build the graph manually to test _calculate_dependency_levels
|
|
graph = {
|
|
self.jet_template_odoo.id: {
|
|
"template": self.jet_template_odoo,
|
|
"name": self.jet_template_odoo.name,
|
|
"reference": self.jet_template_odoo.reference,
|
|
"dependencies": [
|
|
{"template_id": self.jet_template_postgres.id},
|
|
{"template_id": self.jet_template_nginx.id},
|
|
],
|
|
"level": 0, # Will be calculated
|
|
},
|
|
self.jet_template_postgres.id: {
|
|
"template": self.jet_template_postgres,
|
|
"name": self.jet_template_postgres.name,
|
|
"reference": self.jet_template_postgres.reference,
|
|
"dependencies": [{"template_id": self.jet_template_docker.id}],
|
|
"level": 0, # Will be calculated
|
|
},
|
|
self.jet_template_docker.id: {
|
|
"template": self.jet_template_docker,
|
|
"name": self.jet_template_docker.name,
|
|
"reference": self.jet_template_docker.reference,
|
|
"dependencies": [{"template_id": self.jet_template_tower_core.id}],
|
|
"level": 0, # Will be calculated
|
|
},
|
|
self.jet_template_tower_core.id: {
|
|
"template": self.jet_template_tower_core,
|
|
"name": self.jet_template_tower_core.name,
|
|
"reference": self.jet_template_tower_core.reference,
|
|
"dependencies": [],
|
|
"level": 0, # Will be calculated
|
|
},
|
|
}
|
|
|
|
# Call _calculate_dependency_levels
|
|
self.jet_template_odoo._calculate_dependency_levels(graph)
|
|
|
|
# Verify levels
|
|
self.assertEqual(
|
|
graph[self.jet_template_odoo.id]["level"],
|
|
0,
|
|
"Odoo should be level 0 (root)",
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_postgres.id]["level"],
|
|
1,
|
|
"Postgres should be level 1",
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_docker.id]["level"], 2, "Docker should be level 2"
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_tower_core.id]["level"],
|
|
3,
|
|
"Tower Core should be level 3",
|
|
)
|
|
|
|
def test_calculate_dependency_levels_branching_dependencies(self):
|
|
"""Test _calculate_dependency_levels with branching dependencies"""
|
|
# Use existing WordPress template with branching dependencies:
|
|
# WordPress -> MariaDB/Nginx -> Docker
|
|
|
|
# Build the graph manually
|
|
graph = {
|
|
self.jet_template_wordpress.id: {
|
|
"template": self.jet_template_wordpress,
|
|
"name": self.jet_template_wordpress.name,
|
|
"reference": self.jet_template_wordpress.reference,
|
|
"dependencies": [
|
|
{"template_id": self.jet_template_mariadb.id},
|
|
{"template_id": self.jet_template_nginx.id},
|
|
],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_mariadb.id: {
|
|
"template": self.jet_template_mariadb,
|
|
"name": self.jet_template_mariadb.name,
|
|
"reference": self.jet_template_mariadb.reference,
|
|
"dependencies": [{"template_id": self.jet_template_docker.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_nginx.id: {
|
|
"template": self.jet_template_nginx,
|
|
"name": self.jet_template_nginx.name,
|
|
"reference": self.jet_template_nginx.reference,
|
|
"dependencies": [{"template_id": self.jet_template_docker.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_docker.id: {
|
|
"template": self.jet_template_docker,
|
|
"name": self.jet_template_docker.name,
|
|
"reference": self.jet_template_docker.reference,
|
|
"dependencies": [{"template_id": self.jet_template_tower_core.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_tower_core.id: {
|
|
"template": self.jet_template_tower_core,
|
|
"name": self.jet_template_tower_core.name,
|
|
"reference": self.jet_template_tower_core.reference,
|
|
"dependencies": [],
|
|
"level": 0,
|
|
},
|
|
}
|
|
|
|
# Call _calculate_dependency_levels
|
|
self.jet_template_wordpress._calculate_dependency_levels(graph)
|
|
|
|
# Verify levels
|
|
self.assertEqual(
|
|
graph[self.jet_template_wordpress.id]["level"],
|
|
0,
|
|
"WordPress should be level 0 (root)",
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_mariadb.id]["level"], 1, "MariaDB should be level 1"
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_nginx.id]["level"], 1, "Nginx should be level 1"
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_docker.id]["level"],
|
|
2,
|
|
"Docker should be level 2 (shortest path from WordPress)",
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_tower_core.id]["level"],
|
|
3,
|
|
"Tower Core should be level 3",
|
|
)
|
|
|
|
def test_calculate_dependency_levels_multiple_paths(self):
|
|
"""Test _calculate_dependency_levels with multiple paths to same template"""
|
|
# Use existing WordPress template with multiple paths
|
|
|
|
# Build the graph manually
|
|
graph = {
|
|
self.jet_template_wordpress.id: {
|
|
"template": self.jet_template_wordpress,
|
|
"name": self.jet_template_wordpress.name,
|
|
"reference": self.jet_template_wordpress.reference,
|
|
"dependencies": [
|
|
{"template_id": self.jet_template_mariadb.id},
|
|
{"template_id": self.jet_template_nginx.id},
|
|
],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_mariadb.id: {
|
|
"template": self.jet_template_mariadb,
|
|
"name": self.jet_template_mariadb.name,
|
|
"reference": self.jet_template_mariadb.reference,
|
|
"dependencies": [{"template_id": self.jet_template_docker.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_nginx.id: {
|
|
"template": self.jet_template_nginx,
|
|
"name": self.jet_template_nginx.name,
|
|
"reference": self.jet_template_nginx.reference,
|
|
"dependencies": [{"template_id": self.jet_template_docker.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_docker.id: {
|
|
"template": self.jet_template_docker,
|
|
"name": self.jet_template_docker.name,
|
|
"reference": self.jet_template_docker.reference,
|
|
"dependencies": [{"template_id": self.jet_template_tower_core.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_tower_core.id: {
|
|
"template": self.jet_template_tower_core,
|
|
"name": self.jet_template_tower_core.name,
|
|
"reference": self.jet_template_tower_core.reference,
|
|
"dependencies": [],
|
|
"level": 0,
|
|
},
|
|
}
|
|
|
|
# Call _calculate_dependency_levels
|
|
self.jet_template_wordpress._calculate_dependency_levels(graph)
|
|
|
|
# Verify levels - Docker should have level 2 (shortest path from WordPress)
|
|
self.assertEqual(
|
|
graph[self.jet_template_wordpress.id]["level"],
|
|
0,
|
|
"WordPress should be level 0 (root)",
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_mariadb.id]["level"], 1, "MariaDB should be level 1"
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_nginx.id]["level"], 1, "Nginx should be level 1"
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_docker.id]["level"],
|
|
2,
|
|
"Docker should be level 2 (shortest path)",
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_tower_core.id]["level"],
|
|
3,
|
|
"Tower Core should be level 3",
|
|
)
|
|
|
|
def test_calculate_dependency_levels_empty_graph(self):
|
|
"""Test _calculate_dependency_levels with empty graph"""
|
|
# pylint: disable=protected-access
|
|
# Use existing Tower Core template
|
|
|
|
# Empty graph
|
|
graph = {}
|
|
|
|
# Call _calculate_dependency_levels - should not raise error
|
|
self.jet_template_tower_core._calculate_dependency_levels(graph)
|
|
|
|
# Graph should remain empty
|
|
self.assertEqual(len(graph), 0, "Empty graph should remain empty")
|
|
|
|
def test_calculate_dependency_levels_single_template(self):
|
|
"""Test _calculate_dependency_levels with single template"""
|
|
# pylint: disable=protected-access
|
|
# Use existing Tower Core template (has no dependencies)
|
|
|
|
# Single template graph
|
|
graph = {
|
|
self.jet_template_tower_core.id: {
|
|
"template": self.jet_template_tower_core,
|
|
"name": self.jet_template_tower_core.name,
|
|
"reference": self.jet_template_tower_core.reference,
|
|
"dependencies": [],
|
|
"level": 0,
|
|
}
|
|
}
|
|
|
|
# Call _calculate_dependency_levels
|
|
self.jet_template_tower_core._calculate_dependency_levels(graph)
|
|
|
|
# Tower Core should be level 0
|
|
self.assertEqual(
|
|
graph[self.jet_template_tower_core.id]["level"],
|
|
0,
|
|
"Single template should be level 0",
|
|
)
|
|
|
|
def test_calculate_dependency_levels_missing_template_in_graph(self):
|
|
"""Test _calculate_dependency_levels with template not in graph"""
|
|
# pylint: disable=protected-access
|
|
# Use existing Odoo template but reference a non-existent template
|
|
|
|
# Graph with Odoo but not the referenced template
|
|
graph = {
|
|
self.jet_template_odoo.id: {
|
|
"template": self.jet_template_odoo,
|
|
"name": self.jet_template_odoo.name,
|
|
"reference": self.jet_template_odoo.reference,
|
|
"dependencies": [{"template_id": 99999}], # Non-existent template ID
|
|
"level": 0,
|
|
}
|
|
}
|
|
|
|
# Call _calculate_dependency_levels - should handle missing template gracefully
|
|
self.jet_template_odoo._calculate_dependency_levels(graph)
|
|
|
|
# Odoo should be level 0
|
|
self.assertEqual(
|
|
graph[self.jet_template_odoo.id]["level"], 0, "Odoo should be level 0"
|
|
)
|
|
|
|
def test_calculate_dependency_levels_complex_hierarchy(self):
|
|
"""Test _calculate_dependency_levels with complex hierarchy"""
|
|
# pylint: disable=protected-access
|
|
# Use existing templates with complex hierarchy
|
|
# This creates a complex hierarchy
|
|
|
|
# Build the graph manually - only include Odoo's actual dependencies
|
|
graph = {
|
|
self.jet_template_odoo.id: {
|
|
"template": self.jet_template_odoo,
|
|
"name": self.jet_template_odoo.name,
|
|
"reference": self.jet_template_odoo.reference,
|
|
"dependencies": [
|
|
{"template_id": self.jet_template_postgres.id},
|
|
{"template_id": self.jet_template_nginx.id},
|
|
],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_postgres.id: {
|
|
"template": self.jet_template_postgres,
|
|
"name": self.jet_template_postgres.name,
|
|
"reference": self.jet_template_postgres.reference,
|
|
"dependencies": [{"template_id": self.jet_template_docker.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_nginx.id: {
|
|
"template": self.jet_template_nginx,
|
|
"name": self.jet_template_nginx.name,
|
|
"reference": self.jet_template_nginx.reference,
|
|
"dependencies": [{"template_id": self.jet_template_docker.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_docker.id: {
|
|
"template": self.jet_template_docker,
|
|
"name": self.jet_template_docker.name,
|
|
"reference": self.jet_template_docker.reference,
|
|
"dependencies": [{"template_id": self.jet_template_tower_core.id}],
|
|
"level": 0,
|
|
},
|
|
self.jet_template_tower_core.id: {
|
|
"template": self.jet_template_tower_core,
|
|
"name": self.jet_template_tower_core.name,
|
|
"reference": self.jet_template_tower_core.reference,
|
|
"dependencies": [],
|
|
"level": 0,
|
|
},
|
|
}
|
|
|
|
# Call _calculate_dependency_levels from Odoo
|
|
self.jet_template_odoo._calculate_dependency_levels(graph)
|
|
|
|
# Verify levels
|
|
self.assertEqual(
|
|
graph[self.jet_template_odoo.id]["level"],
|
|
0,
|
|
"Odoo should be level 0 (root)",
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_postgres.id]["level"],
|
|
1,
|
|
"Postgres should be level 1",
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_nginx.id]["level"], 1, "Nginx should be level 1"
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_docker.id]["level"], 2, "Docker should be level 2"
|
|
)
|
|
self.assertEqual(
|
|
graph[self.jet_template_tower_core.id]["level"],
|
|
3,
|
|
"Tower Core should be level 3",
|
|
)
|
|
|
|
# Verify that only Odoo's dependencies are in the graph
|
|
expected_template_ids = [
|
|
self.jet_template_odoo.id,
|
|
self.jet_template_postgres.id,
|
|
self.jet_template_nginx.id,
|
|
self.jet_template_docker.id,
|
|
self.jet_template_tower_core.id,
|
|
]
|
|
self.assertEqual(
|
|
set(graph.keys()),
|
|
set(expected_template_ids),
|
|
"Graph should only contain Odoo's dependencies",
|
|
)
|
|
|
|
def test_get_all_dependencies_simple_chain(self):
|
|
"""Test _get_all_dependencies with simple dependency chain"""
|
|
# pylint: disable=protected-access
|
|
# Use existing Odoo dependency chain:
|
|
# Odoo -> Postgres/Nginx -> Docker -> Tower Core
|
|
|
|
dependencies = self.jet_template_odoo._get_all_dependencies()
|
|
|
|
# Should return all dependencies in level order (closest first)
|
|
expected_dependencies = {
|
|
self.jet_template_postgres,
|
|
self.jet_template_nginx,
|
|
self.jet_template_docker,
|
|
self.jet_template_tower_core,
|
|
}
|
|
self.assertEqual(
|
|
set(dependencies),
|
|
expected_dependencies,
|
|
"Should return all expected dependencies",
|
|
)
|
|
|
|
# Verify the order is correct (level 1, then level 2, then level 3)
|
|
# Postgres and Nginx should be first (level 1)
|
|
self.assertIn(
|
|
self.jet_template_postgres,
|
|
dependencies[:2],
|
|
"Postgres should be in first two dependencies",
|
|
)
|
|
self.assertIn(
|
|
self.jet_template_nginx,
|
|
dependencies[:2],
|
|
"Nginx should be in first two dependencies",
|
|
)
|
|
|
|
# Docker should be third (level 2)
|
|
self.assertEqual(
|
|
dependencies[2], self.jet_template_docker, "Docker should be third"
|
|
)
|
|
|
|
# Tower Core should be last (level 3)
|
|
self.assertEqual(
|
|
dependencies[3], self.jet_template_tower_core, "Tower Core should be last"
|
|
)
|
|
|
|
def test_get_all_dependencies_no_dependencies(self):
|
|
"""Test _get_all_dependencies with template that has no dependencies"""
|
|
# pylint: disable=protected-access
|
|
# Use Tower Core which has no dependencies
|
|
|
|
dependencies = self.jet_template_tower_core._get_all_dependencies()
|
|
|
|
# Should return empty list
|
|
self.assertEqual(
|
|
dependencies,
|
|
[],
|
|
"Should return empty list for template with no dependencies",
|
|
)
|
|
|
|
def test_get_all_dependencies_wordpress_chain(self):
|
|
"""Test _get_all_dependencies with WordPress dependency chain"""
|
|
# pylint: disable=protected-access
|
|
# Use WordPress dependency chain:
|
|
# WordPress -> MariaDB/Nginx -> Docker -> Tower Core
|
|
|
|
dependencies = self.jet_template_wordpress._get_all_dependencies()
|
|
|
|
# Should return all dependencies in level order
|
|
expected_dependencies = {
|
|
self.jet_template_mariadb,
|
|
self.jet_template_nginx,
|
|
self.jet_template_docker,
|
|
self.jet_template_tower_core,
|
|
}
|
|
self.assertEqual(
|
|
set(dependencies),
|
|
expected_dependencies,
|
|
"Should return all expected dependencies",
|
|
)
|
|
|
|
# Verify the order is correct
|
|
# MariaDB and Nginx should be first (level 1)
|
|
self.assertIn(
|
|
self.jet_template_mariadb,
|
|
dependencies[:2],
|
|
"MariaDB should be in first two dependencies",
|
|
)
|
|
self.assertIn(
|
|
self.jet_template_nginx,
|
|
dependencies[:2],
|
|
"Nginx should be in first two dependencies",
|
|
)
|
|
|
|
# Docker should be third (level 2)
|
|
self.assertEqual(
|
|
dependencies[2], self.jet_template_docker, "Docker should be third"
|
|
)
|
|
|
|
# Tower Core should be last (level 3)
|
|
self.assertEqual(
|
|
dependencies[3], self.jet_template_tower_core, "Tower Core should be last"
|
|
)
|
|
|
|
def test_get_all_dependencies_docker_chain(self):
|
|
"""Test _get_all_dependencies with Docker dependency chain"""
|
|
# pylint: disable=protected-access
|
|
# Use Docker dependency chain: Docker -> Tower Core
|
|
|
|
dependencies = self.jet_template_docker._get_all_dependencies()
|
|
|
|
# Should return only Tower Core
|
|
expected_dependencies = [self.jet_template_tower_core]
|
|
self.assertEqual(
|
|
dependencies, expected_dependencies, "Should return only Tower Core"
|
|
)
|
|
|
|
def test_get_all_dependencies_nginx_chain(self):
|
|
"""Test _get_all_dependencies with Nginx dependency chain"""
|
|
# pylint: disable=protected-access
|
|
# Use Nginx dependency chain: Nginx -> Docker -> Tower Core
|
|
|
|
dependencies = self.jet_template_nginx._get_all_dependencies()
|
|
|
|
# Should return Docker and Tower Core
|
|
expected_dependencies = [self.jet_template_docker, self.jet_template_tower_core]
|
|
self.assertEqual(
|
|
dependencies, expected_dependencies, "Should return Docker and Tower Core"
|
|
)
|
|
|
|
def test_get_all_dependencies_complex_scenario(self):
|
|
"""Test _get_all_dependencies with complex dependency scenario"""
|
|
# pylint: disable=protected-access
|
|
# Use existing WooCommerce with Odoo template
|
|
# This tests the scenario where a template has multiple dependency paths
|
|
|
|
dependencies = self.jet_template_woocommerce_odoo._get_all_dependencies()
|
|
|
|
# Should include all dependencies from both Odoo and WordPress
|
|
# Expected: Odoo, WordPress, Postgres, MariaDB, Nginx, Docker, Tower Core
|
|
expected_template_ids = [
|
|
self.jet_template_odoo.id,
|
|
self.jet_template_wordpress.id,
|
|
self.jet_template_postgres.id,
|
|
self.jet_template_mariadb.id,
|
|
self.jet_template_nginx.id,
|
|
self.jet_template_docker.id,
|
|
self.jet_template_tower_core.id,
|
|
]
|
|
|
|
actual_template_ids = [dep.id for dep in dependencies]
|
|
self.assertEqual(
|
|
set(actual_template_ids),
|
|
set(expected_template_ids),
|
|
"Should include all dependencies from both Odoo and WordPress",
|
|
)
|
|
|
|
# Verify that dependencies are ordered by level
|
|
# Level 1: Odoo, WordPress
|
|
# Level 2: Postgres, MariaDB, Nginx
|
|
# Level 3: Docker
|
|
# Level 4: Tower Core
|
|
|
|
# Check that Odoo and WordPress are in the first two positions
|
|
self.assertIn(
|
|
self.jet_template_odoo,
|
|
dependencies[:2],
|
|
"Odoo should be in first two dependencies",
|
|
)
|
|
self.assertIn(
|
|
self.jet_template_wordpress,
|
|
dependencies[:2],
|
|
"WordPress should be in first two dependencies",
|
|
)
|
|
|
|
# Check that Tower Core is last
|
|
self.assertEqual(
|
|
dependencies[-1], self.jet_template_tower_core, "Tower Core should be last"
|
|
)
|
|
|
|
def test_get_all_dependencies_excludes_self(self):
|
|
"""Test _get_all_dependencies excludes the template itself"""
|
|
# pylint: disable=protected-access
|
|
# Use Odoo template
|
|
|
|
dependencies = self.jet_template_odoo._get_all_dependencies()
|
|
|
|
# Should not include Odoo itself
|
|
self.assertNotIn(
|
|
self.jet_template_odoo,
|
|
dependencies,
|
|
"Should not include the template itself",
|
|
)
|
|
|
|
# Verify all returned dependencies are different from the root template
|
|
for dependency in dependencies:
|
|
self.assertNotEqual(
|
|
dependency.id,
|
|
self.jet_template_odoo.id,
|
|
f"Should not include template with ID {dependency.id}",
|
|
)
|
|
|
|
def test_get_all_dependencies_same_level_must_order_transitive_edges(self):
|
|
"""
|
|
If root A depends on B and C directly, and C also depends on B, then B and
|
|
C share the same shortest-path level. Install lines use reverse order by
|
|
``order``; the dependency list must place C before B so B gets a higher
|
|
line order and is installed before C.
|
|
"""
|
|
# pylint: disable=protected-access
|
|
tpl_b = self.JetTemplate.create(
|
|
{
|
|
"name": "Topo Base B",
|
|
"reference": "topo_base_b",
|
|
}
|
|
)
|
|
tpl_c = self.JetTemplate.create(
|
|
{
|
|
"name": "Topo Mid C",
|
|
"reference": "topo_mid_c",
|
|
}
|
|
)
|
|
tpl_a = self.JetTemplate.create(
|
|
{
|
|
"name": "Topo Root A",
|
|
"reference": "topo_root_a",
|
|
}
|
|
)
|
|
# C depends on B
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": tpl_c.id,
|
|
"template_required_id": tpl_b.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
# A depends on C first, then B so graph traversal tends to visit B before C
|
|
# in ``graph.items()`` while both stay at level 1.
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": tpl_a.id,
|
|
"template_required_id": tpl_c.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
self.JetTemplateDependency.create(
|
|
{
|
|
"template_id": tpl_a.id,
|
|
"template_required_id": tpl_b.id,
|
|
"state_required_id": self.state_running.id,
|
|
}
|
|
)
|
|
|
|
dependencies = tpl_a._get_all_dependencies()
|
|
idx_b = next(i for i, t in enumerate(dependencies) if t.id == tpl_b.id)
|
|
idx_c = next(i for i, t in enumerate(dependencies) if t.id == tpl_c.id)
|
|
|
|
self.assertLess(
|
|
idx_c,
|
|
idx_b,
|
|
"C must appear before B in dependency order so install (reverse order)"
|
|
" runs B before C when C depends on B",
|
|
)
|
|
|
|
def test_get_all_dependencies_consistency_with_build_graph(self):
|
|
"""
|
|
_get_all_dependencies must return dependents before their prerequisites.
|
|
|
|
Correctness is verified against the graph edges directly (the topological
|
|
invariant) rather than re-running _topological_sort_dependency_graph, which
|
|
would create a circular check where a bug in the sort masks itself.
|
|
"""
|
|
# pylint: disable=protected-access
|
|
graph = self.jet_template_odoo._build_dependency_graph()
|
|
dependencies = self.jet_template_odoo._get_all_dependencies()
|
|
|
|
self.assertTrue(dependencies, "Expected a non-empty dependency list")
|
|
|
|
index = {tmpl.id: i for i, tmpl in enumerate(dependencies)}
|
|
|
|
for u_id, info in graph.items():
|
|
if u_id not in index:
|
|
continue
|
|
for dep in info["dependencies"]:
|
|
v_id = dep["template_id"]
|
|
if v_id not in index:
|
|
continue
|
|
self.assertLess(
|
|
index[u_id],
|
|
index[v_id],
|
|
f"{graph[u_id]['name']} (dependent) must appear before "
|
|
f"{graph[v_id]['name']} (prerequisite)",
|
|
)
|
|
|
|
def test_get_all_dependencies_woocommerce_odoo_chain(self):
|
|
"""Test _get_all_dependencies with WooCommerce with Odoo dependency chain"""
|
|
# pylint: disable=protected-access
|
|
# Use WooCommerce with Odoo dependency chain:
|
|
# WooCommerce -> WordPress/Odoo ->
|
|
# MariaDB/Postgres/Nginx -> Docker -> Tower Core
|
|
|
|
dependencies = self.jet_template_woocommerce_odoo._get_all_dependencies()
|
|
|
|
# Should include all dependencies from both WordPress and Odoo
|
|
# Expected: WordPress, Odoo, MariaDB, Postgres, Nginx, Docker, Tower Core
|
|
expected_template_ids = [
|
|
self.jet_template_wordpress.id,
|
|
self.jet_template_odoo.id,
|
|
self.jet_template_mariadb.id,
|
|
self.jet_template_postgres.id,
|
|
self.jet_template_nginx.id,
|
|
self.jet_template_docker.id,
|
|
self.jet_template_tower_core.id,
|
|
]
|
|
|
|
actual_template_ids = [dep.id for dep in dependencies]
|
|
self.assertEqual(
|
|
set(actual_template_ids),
|
|
set(expected_template_ids),
|
|
"Should include all dependencies from both WordPress and Odoo",
|
|
)
|
|
|
|
# Verify that dependencies are ordered by level
|
|
# Level 1: WordPress, Odoo
|
|
# Level 2: MariaDB, Postgres, Nginx
|
|
# Level 3: Docker
|
|
# Level 4: Tower Core
|
|
|
|
# Check that WordPress and Odoo are in the first two positions
|
|
self.assertIn(
|
|
self.jet_template_wordpress,
|
|
dependencies[:2],
|
|
"WordPress should be in first two dependencies",
|
|
)
|
|
self.assertIn(
|
|
self.jet_template_odoo,
|
|
dependencies[:2],
|
|
"Odoo should be in first two dependencies",
|
|
)
|
|
|
|
# Check that Tower Core is last
|
|
self.assertEqual(
|
|
dependencies[-1], self.jet_template_tower_core, "Tower Core should be last"
|
|
)
|
|
|
|
# Verify that all level 2 dependencies are present
|
|
level_2_deps = [
|
|
self.jet_template_mariadb,
|
|
self.jet_template_postgres,
|
|
self.jet_template_nginx,
|
|
]
|
|
for dep in level_2_deps:
|
|
self.assertIn(dep, dependencies, f"{dep.name} should be in dependencies")
|
|
|
|
# Verify that Docker is present
|
|
self.assertIn(
|
|
self.jet_template_docker, dependencies, "Docker should be in dependencies"
|
|
)
|
|
|
|
def test_get_action_path_with_destroy_action_only(self):
|
|
"""
|
|
Test _get_action_path with only destroy action set
|
|
"""
|
|
# Create states
|
|
state_running = self.JetState.create(
|
|
{
|
|
"name": "Running",
|
|
"reference": "running",
|
|
"sequence": 20,
|
|
}
|
|
)
|
|
state_stopped = self.JetState.create(
|
|
{
|
|
"name": "Stopped",
|
|
"reference": "stopped",
|
|
"sequence": 30,
|
|
}
|
|
)
|
|
|
|
# Create destroy action
|
|
destroy_action = self.JetAction.create(
|
|
{
|
|
"name": "Destroy Action",
|
|
"reference": "destroy_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": state_running.id,
|
|
"state_to_id": False,
|
|
"state_transit_id": state_stopped.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Set destroy action
|
|
self.clean_template.action_destroy_id = destroy_action
|
|
|
|
# Test path without state_to (should use destroy action)
|
|
result = self.clean_template._get_action_path(state_from=state_running)
|
|
self.assertEqual(
|
|
result,
|
|
[destroy_action],
|
|
"Should return destroy action when no state_to provided",
|
|
)
|
|
|
|
# Test path with state_to (should not use destroy action)
|
|
result = self.clean_template._get_action_path(
|
|
state_from=state_running, state_to=state_stopped
|
|
)
|
|
self.assertEqual(
|
|
result,
|
|
[],
|
|
"Should return empty list when state_to provided and no path exists",
|
|
)
|
|
|
|
def test_get_action_path_same_state(self):
|
|
"""
|
|
Test _get_action_path when start and end states are the same
|
|
"""
|
|
# Test same state without destroy action
|
|
result = self.clean_template._get_action_path(
|
|
state_from=self.state_a, state_to=self.state_a
|
|
)
|
|
self.assertEqual(
|
|
result, [], "Should return empty list for same start and end state"
|
|
)
|
|
|
|
# Create destroy action
|
|
destroy_action = self.JetAction.create(
|
|
{
|
|
"name": "Destroy Action",
|
|
"reference": "destroy_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": False,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
self.clean_template.action_destroy_id = destroy_action
|
|
|
|
# Test same state with destroy action (no state_to provided)
|
|
result = self.clean_template._get_action_path(state_from=self.state_a)
|
|
self.assertEqual(
|
|
result,
|
|
[destroy_action],
|
|
"Should return destroy action for same state when no state_to provided",
|
|
)
|
|
|
|
def test_get_action_path_direct_path(self):
|
|
"""
|
|
Test _get_action_path with direct path between states
|
|
"""
|
|
# Create direct action
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Test direct path
|
|
result = self.clean_template._get_action_path(
|
|
state_from=self.state_a, state_to=self.state_b
|
|
)
|
|
self.assertEqual(result, [action_ab], "Should return direct action path")
|
|
|
|
def test_get_action_path_multi_step_path(self):
|
|
"""
|
|
Test _get_action_path with multi-step path
|
|
"""
|
|
# Create multi-step actions
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_bc = self.JetAction.create(
|
|
{
|
|
"name": "Action B to C",
|
|
"reference": "action_bc",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Test multi-step path
|
|
result = self.clean_template._get_action_path(
|
|
state_from=self.state_a, state_to=self.state_c
|
|
)
|
|
expected_path = [action_ab, action_bc]
|
|
self.assertEqual(result, expected_path, "Should return multi-step action path")
|
|
|
|
def test_get_action_path_with_create_and_multi_step(self):
|
|
"""
|
|
Test _get_action_path with create action and multi-step path
|
|
"""
|
|
# Create create action
|
|
create_action = self.JetAction.create(
|
|
{
|
|
"name": "Create Action",
|
|
"reference": "create_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": False,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_a.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Create transition action
|
|
action_rs = self.JetAction.create(
|
|
{
|
|
"name": "Action Running to Stopped",
|
|
"reference": "action_rs",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_c.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Set create action
|
|
self.clean_template.action_create_id = create_action
|
|
|
|
# Test path from create to final state
|
|
result = self.clean_template._get_action_path(state_to=self.state_c)
|
|
expected_path = [create_action, action_rs]
|
|
self.assertEqual(
|
|
result, expected_path, "Should return create action + transition path"
|
|
)
|
|
|
|
def test_get_action_path_with_multi_step_and_destroy(self):
|
|
"""
|
|
Test _get_action_path with multi-step path and destroy action
|
|
"""
|
|
# Create multi-step actions
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_bc = self.JetAction.create(
|
|
{
|
|
"name": "Action B to C",
|
|
"reference": "action_bc",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Create destroy action
|
|
destroy_action = self.JetAction.create(
|
|
{
|
|
"name": "Destroy Action",
|
|
"reference": "destroy_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_c.id,
|
|
"state_to_id": False,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Set destroy action
|
|
self.clean_template.action_destroy_id = destroy_action
|
|
|
|
# Test path from A to destroy
|
|
result = self.clean_template._get_action_path(state_from=self.state_a)
|
|
expected_path = [action_ab, action_bc, destroy_action]
|
|
self.assertEqual(
|
|
result, expected_path, "Should return multi-step path + destroy action"
|
|
)
|
|
|
|
def test_get_action_path_complete_lifecycle(self):
|
|
"""
|
|
Test _get_action_path with complete lifecycle (create -> multi-step -> destroy)
|
|
"""
|
|
# Create create action
|
|
create_action = self.JetAction.create(
|
|
{
|
|
"name": "Create Action",
|
|
"reference": "create_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": False,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_a.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Create transition action
|
|
action_rs = self.JetAction.create(
|
|
{
|
|
"name": "Action Running to Stopped",
|
|
"reference": "action_rs",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_c.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Create destroy action
|
|
destroy_action = self.JetAction.create(
|
|
{
|
|
"name": "Destroy Action",
|
|
"reference": "destroy_action",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_c.id,
|
|
"state_to_id": False,
|
|
"state_transit_id": self.state_c.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Set border actions
|
|
self.clean_template.action_create_id = create_action
|
|
self.clean_template.action_destroy_id = destroy_action
|
|
|
|
# Test complete lifecycle
|
|
result = self.clean_template._get_action_path()
|
|
expected_path = [create_action, action_rs, destroy_action]
|
|
self.assertEqual(result, expected_path, "Should return complete lifecycle path")
|
|
|
|
def test_get_action_path_no_path_exists(self):
|
|
"""
|
|
Test _get_action_path when no path exists between states
|
|
"""
|
|
# Create action that doesn't connect A to C
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Action B to C",
|
|
"reference": "action_bc",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Test path from A to C (no path exists)
|
|
result = self.clean_template._get_action_path(
|
|
state_from=self.state_a, state_to=self.state_c
|
|
)
|
|
self.assertEqual(result, [], "Should return empty list when no path exists")
|
|
|
|
def test_get_action_path_complex_multi_level_path(self):
|
|
"""
|
|
Test _get_action_path with complex multi-level path
|
|
"""
|
|
# Create additional states for this test
|
|
state_e = self.JetState.create(
|
|
{
|
|
"name": "State E",
|
|
"reference": "state_e",
|
|
"sequence": 50,
|
|
}
|
|
)
|
|
|
|
# Create complex multi-level actions
|
|
action_ab = self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_bc = self.JetAction.create(
|
|
{
|
|
"name": "Action B to C",
|
|
"reference": "action_bc",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_cd = self.JetAction.create(
|
|
{
|
|
"name": "Action C to D",
|
|
"reference": "action_cd",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_c.id,
|
|
"state_to_id": self.state_d.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
action_de = self.JetAction.create(
|
|
{
|
|
"name": "Action D to E",
|
|
"reference": "action_de",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_d.id,
|
|
"state_to_id": state_e.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Test complex multi-level path
|
|
result = self.clean_template._get_action_path(
|
|
state_from=self.state_a, state_to=state_e
|
|
)
|
|
expected_path = [action_ab, action_bc, action_cd, action_de]
|
|
self.assertEqual(
|
|
result, expected_path, "Should return complex multi-level path"
|
|
)
|
|
|
|
def test_get_action_path_shortest_path_selection(self):
|
|
"""
|
|
Test _get_action_path selects shortest path when multiple paths exist
|
|
"""
|
|
# Create short path: A -> C
|
|
action_ac = self.JetAction.create(
|
|
{
|
|
"name": "Action A to C (short)",
|
|
"reference": "action_ac",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Create long path: A -> B -> D -> C
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Action A to B",
|
|
"reference": "action_ab",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_a.id,
|
|
"state_to_id": self.state_b.id,
|
|
"state_transit_id": self.state_starting.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Action B to D",
|
|
"reference": "action_bd",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_b.id,
|
|
"state_to_id": self.state_d.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
self.JetAction.create(
|
|
{
|
|
"name": "Action D to C",
|
|
"reference": "action_dc",
|
|
"jet_template_id": self.clean_template.id,
|
|
"state_from_id": self.state_d.id,
|
|
"state_to_id": self.state_c.id,
|
|
"state_transit_id": self.state_stopping.id,
|
|
"priority": 10,
|
|
}
|
|
)
|
|
|
|
# Test that shortest path is selected
|
|
result = self.clean_template._get_action_path(
|
|
state_from=self.state_a, state_to=self.state_c
|
|
)
|
|
expected_path = [action_ac] # Shortest path
|
|
self.assertEqual(
|
|
result,
|
|
expected_path,
|
|
"Should select shortest path when multiple paths exist",
|
|
)
|
|
|
|
def test_check_dependency_satisfaction_no_dependencies(self):
|
|
"""Test _check_dependency_satisfaction when template has no dependencies"""
|
|
# pylint: disable=protected-access
|
|
server = self.server_test_1
|
|
|
|
# Test with template that has no dependencies
|
|
missing_templates = self.jet_template_tower_core._check_dependency_satisfaction(
|
|
server
|
|
)
|
|
|
|
# Should return empty list since tower_core has no dependencies
|
|
self.assertEqual(
|
|
len(missing_templates),
|
|
0,
|
|
"Should return empty list when no dependencies exist",
|
|
)
|
|
|
|
def test_check_dependency_satisfaction_all_missing(self):
|
|
"""Test _check_dependency_satisfaction when all dependencies are missing"""
|
|
# pylint: disable=protected-access
|
|
server = self.server_test_1
|
|
|
|
# Test with different templates that have dependencies
|
|
templates_to_test = [
|
|
self.jet_template_nginx,
|
|
self.jet_template_odoo,
|
|
self.jet_template_woocommerce_odoo,
|
|
]
|
|
|
|
for template in templates_to_test:
|
|
# Get actual dependencies for template
|
|
all_deps = template._get_all_dependencies()
|
|
|
|
# Test - should return all missing dependencies
|
|
missing_templates = template._check_dependency_satisfaction(server)
|
|
|
|
# Should return all dependencies since none are installed
|
|
expected_dependencies = set(all_deps)
|
|
actual_dependencies = set(missing_templates)
|
|
self.assertEqual(
|
|
actual_dependencies,
|
|
expected_dependencies,
|
|
f"Should return all missing dependencies for {template.name}",
|
|
)
|
|
|
|
def test_check_dependency_satisfaction_all_satisfied(self):
|
|
"""Test _check_dependency_satisfaction when all dependencies are satisfied"""
|
|
# pylint: disable=protected-access
|
|
server = self.server_test_1
|
|
|
|
# Test with different templates that have dependencies
|
|
templates_to_test = [
|
|
self.jet_template_nginx,
|
|
self.jet_template_odoo,
|
|
self.jet_template_woocommerce_odoo,
|
|
]
|
|
|
|
for template in templates_to_test:
|
|
# Install all dependencies for this template
|
|
all_deps = template._get_all_dependencies()
|
|
for dep_template in all_deps:
|
|
dep_template.server_ids = [(4, server.id)]
|
|
|
|
# Test - should return empty list
|
|
missing_templates = template._check_dependency_satisfaction(server)
|
|
|
|
# Should return empty list since all dependencies are now installed
|
|
self.assertEqual(
|
|
len(missing_templates),
|
|
0,
|
|
f"Should return empty list for {template.name}",
|
|
)
|
|
|
|
def test_check_dependency_satisfaction_partial_installation(self):
|
|
"""Test _check_dependency_satisfaction with partial installation"""
|
|
# pylint: disable=protected-access
|
|
server = self.server_test_1
|
|
|
|
# Get all dependencies for odoo
|
|
all_deps = self.jet_template_odoo._get_all_dependencies()
|
|
|
|
# Install some dependencies but not all (install first half)
|
|
half_count = len(all_deps) // 2
|
|
for i, dep in enumerate(all_deps):
|
|
if i < half_count:
|
|
dep.server_ids = [(4, server.id)]
|
|
|
|
# Test with odoo
|
|
missing_templates = self.jet_template_odoo._check_dependency_satisfaction(
|
|
server
|
|
)
|
|
|
|
# Should return the remaining uninstalled dependencies
|
|
expected_missing = set(all_deps[half_count:])
|
|
actual_missing = set(missing_templates)
|
|
self.assertEqual(
|
|
actual_missing,
|
|
expected_missing,
|
|
"Should return only the missing dependencies",
|
|
)
|
|
|
|
def test_check_dependency_satisfaction_no_server(self):
|
|
"""Test _check_dependency_satisfaction when server is None"""
|
|
# pylint: disable=protected-access
|
|
|
|
# Test with odoo and None server
|
|
missing_templates = self.jet_template_odoo._check_dependency_satisfaction(None)
|
|
|
|
# Should return empty list when server is None (no server to check against)
|
|
self.assertEqual(
|
|
len(missing_templates), 0, "Should return empty list when server is None"
|
|
)
|
|
|
|
def test_check_dependency_satisfaction_multiple_servers(self):
|
|
"""Test _check_dependency_satisfaction with different server states"""
|
|
# pylint: disable=protected-access
|
|
server1 = self.server_test_1
|
|
server2 = self.server_test_2
|
|
|
|
# Get actual dependencies for nginx
|
|
all_deps = self.jet_template_nginx._get_all_dependencies()
|
|
|
|
# Install all dependencies on server1
|
|
for dep in all_deps:
|
|
dep.server_ids = [(4, server1.id)]
|
|
|
|
# Test with nginx on both servers
|
|
missing_templates_server1 = (
|
|
self.jet_template_nginx._check_dependency_satisfaction(server1)
|
|
)
|
|
missing_templates_server2 = (
|
|
self.jet_template_nginx._check_dependency_satisfaction(server2)
|
|
)
|
|
|
|
# Server1 should have no missing dependencies
|
|
self.assertEqual(
|
|
len(missing_templates_server1),
|
|
0,
|
|
"Server1 should have no missing dependencies",
|
|
)
|
|
self.assertEqual(
|
|
len(missing_templates_server2),
|
|
len(all_deps),
|
|
"Server2 should have all dependencies missing",
|
|
)
|
|
|
|
# Verify server2 has all the expected missing dependencies
|
|
expected_missing_server2 = set(all_deps)
|
|
actual_missing_server2 = set(missing_templates_server2)
|
|
self.assertEqual(
|
|
actual_missing_server2,
|
|
expected_missing_server2,
|
|
"Server2 should be missing all dependencies",
|
|
)
|
|
|
|
def test_check_dependency_satisfaction_self_dependency(self):
|
|
"""Test _check_dependency_satisfaction with template that depends on itself"""
|
|
# pylint: disable=protected-access
|
|
server = self.server_test_1
|
|
|
|
# Create a template that depends on itself
|
|
# But let's test the method behavior anyway
|
|
self_loop_template = self.JetTemplate.create(
|
|
{
|
|
"name": "Self Loop Template",
|
|
"reference": "self_loop_template",
|
|
}
|
|
)
|
|
|
|
# Manually create a dependency record (this would normally be prevented)
|
|
# We'll test the method's behavior when it encounters this situation
|
|
missing_templates = self_loop_template._check_dependency_satisfaction(server)
|
|
|
|
# Should return empty list since template has no dependencies
|
|
self.assertEqual(
|
|
len(missing_templates),
|
|
0,
|
|
"Should return empty list for template with no dependencies",
|
|
)
|
|
|
|
def test_get_all_depend_on_this_no_dependents(self):
|
|
"""Test _get_all_depend_on_this when template has no dependents"""
|
|
# pylint: disable=protected-access
|
|
|
|
# Test with woocommerce_odoo which should have no dependents
|
|
dependents = self.jet_template_woocommerce_odoo._get_all_depend_on_this()
|
|
|
|
# Should return empty recordset since no templates depend on woocommerce_odoo
|
|
self.assertEqual(
|
|
len(dependents),
|
|
0,
|
|
"Should return empty recordset when no templates depend on this one",
|
|
)
|
|
|
|
def test_get_all_depend_on_this_docker_dependents(self):
|
|
"""Test _get_all_depend_on_this with docker's dependents"""
|
|
# pylint: disable=protected-access
|
|
|
|
# Test with docker - should have all dependents (direct and indirect)
|
|
dependents = self.jet_template_docker._get_all_depend_on_this()
|
|
|
|
# Should return all templates that depend on docker (directly or indirectly)
|
|
# docker -> nginx/postgres/mariadb -> odoo/wordpress -> woocommerce_odoo
|
|
expected_dependents = {
|
|
self.jet_template_nginx,
|
|
self.jet_template_postgres,
|
|
self.jet_template_mariadb,
|
|
self.jet_template_odoo,
|
|
self.jet_template_wordpress,
|
|
self.jet_template_woocommerce_odoo,
|
|
}
|
|
actual_dependents = set(dependents)
|
|
|
|
# Filter out any templates that aren't in the expected set
|
|
# (some tests might have created additional dependencies)
|
|
actual_dependents_filtered = {
|
|
t for t in actual_dependents if t in expected_dependents
|
|
}
|
|
|
|
self.assertEqual(
|
|
actual_dependents_filtered,
|
|
expected_dependents,
|
|
"Should return all dependents of docker",
|
|
)
|
|
|
|
def test_get_all_depend_on_this_indirect_dependents(self):
|
|
"""Test _get_all_depend_on_this with indirect dependents"""
|
|
# pylint: disable=protected-access
|
|
|
|
# Test with tower_core - should have many indirect dependents
|
|
dependents = self.jet_template_tower_core._get_all_depend_on_this()
|
|
|
|
# Should return all templates that depend on tower_core (directly or indirectly)
|
|
# tower_core -> docker -> nginx/postgres -> odoo/wordpress -> woocommerce_odoo
|
|
expected_dependents = {
|
|
self.jet_template_docker,
|
|
self.jet_template_nginx,
|
|
self.jet_template_postgres,
|
|
self.jet_template_mariadb,
|
|
self.jet_template_odoo,
|
|
self.jet_template_wordpress,
|
|
self.jet_template_woocommerce_odoo,
|
|
}
|
|
actual_dependents = set(dependents)
|
|
|
|
# Filter out any templates that aren't in the expected set
|
|
# (some tests might have created additional dependencies)
|
|
actual_dependents_filtered = {
|
|
t for t in actual_dependents if t in expected_dependents
|
|
}
|
|
|
|
self.assertEqual(
|
|
actual_dependents_filtered,
|
|
expected_dependents,
|
|
"Should return all dependents including indirect ones",
|
|
)
|
|
|
|
def test_get_all_depend_on_this_complex_hierarchy(self):
|
|
"""Test _get_all_depend_on_this with complex dependency hierarchy"""
|
|
# pylint: disable=protected-access
|
|
|
|
# Test with nginx - should have odoo, wordpress, and woocommerce_odoo
|
|
dependents = self.jet_template_nginx._get_all_depend_on_this()
|
|
|
|
# Should return odoo, wordpress, and woocommerce_odoo
|
|
expected_dependents = {
|
|
self.jet_template_odoo,
|
|
self.jet_template_wordpress,
|
|
self.jet_template_woocommerce_odoo,
|
|
}
|
|
actual_dependents = set(dependents)
|
|
|
|
# Filter out any templates that aren't in the expected set
|
|
# (some tests might have created additional dependencies)
|
|
actual_dependents_filtered = {
|
|
t for t in actual_dependents if t in expected_dependents
|
|
}
|
|
|
|
self.assertEqual(
|
|
actual_dependents_filtered,
|
|
expected_dependents,
|
|
"Should return all dependents in complex hierarchy",
|
|
)
|
|
|
|
def test_get_all_depend_on_this_multiple_levels(self):
|
|
"""Test _get_all_depend_on_this with multiple dependency levels"""
|
|
# pylint: disable=protected-access
|
|
|
|
# Test with postgres - should have odoo and woocommerce_odoo as dependents
|
|
dependents = self.jet_template_postgres._get_all_depend_on_this()
|
|
|
|
# Should return odoo and woocommerce_odoo
|
|
expected_dependents = {
|
|
self.jet_template_odoo,
|
|
self.jet_template_woocommerce_odoo,
|
|
}
|
|
actual_dependents = set(dependents)
|
|
|
|
# Filter out any templates that aren't in the expected set
|
|
# (some tests might have created additional dependencies)
|
|
actual_dependents_filtered = {
|
|
t for t in actual_dependents if t in expected_dependents
|
|
}
|
|
|
|
self.assertEqual(
|
|
actual_dependents_filtered,
|
|
expected_dependents,
|
|
"Should return dependents across multiple levels",
|
|
)
|
|
|
|
def test_get_all_depend_on_this_self_dependency(self):
|
|
"""Test _get_all_depend_on_this with template that has no dependents"""
|
|
# pylint: disable=protected-access
|
|
|
|
# Test with a template that has no dependents
|
|
dependents = self.jet_template_woocommerce_odoo._get_all_depend_on_this()
|
|
|
|
# Should return empty recordset
|
|
self.assertEqual(
|
|
len(dependents),
|
|
0,
|
|
"Should return empty recordset for template with no dependents",
|
|
)
|
|
|
|
def test_get_all_depend_on_this_consistency_with_dependencies(self):
|
|
"""Test that _get_all_depend_on_this is consistent with _get_all_dependencies"""
|
|
# pylint: disable=protected-access
|
|
|
|
# For each template, check that its dependents are consistent
|
|
templates_to_test = [
|
|
self.jet_template_tower_core,
|
|
self.jet_template_docker,
|
|
self.jet_template_nginx,
|
|
self.jet_template_postgres,
|
|
self.jet_template_odoo,
|
|
]
|
|
|
|
for template in templates_to_test:
|
|
# Get templates that depend on this template
|
|
dependents = template._get_all_depend_on_this()
|
|
|
|
# For each dependent, check that this template is in its dependencies
|
|
for dependent in dependents:
|
|
dependent_deps = dependent._get_all_dependencies()
|
|
self.assertIn(
|
|
template,
|
|
dependent_deps,
|
|
f"{dependent.name} should have {template.name} in its dependencies",
|
|
)
|
|
|
|
def test_get_all_depend_on_this_circular_dependency_handling(self):
|
|
"""Test _get_all_depend_on_this handles circular dependencies correctly"""
|
|
# pylint: disable=protected-access
|
|
|
|
# Test with templates that might have circular dependencies
|
|
# This test ensures the method doesn't get stuck in infinite loops
|
|
templates_to_test = [
|
|
self.jet_template_tower_core,
|
|
self.jet_template_docker,
|
|
self.jet_template_nginx,
|
|
self.jet_template_postgres,
|
|
self.jet_template_odoo,
|
|
]
|
|
|
|
for template in templates_to_test:
|
|
# This should not raise an exception or get stuck
|
|
dependents = template._get_all_depend_on_this()
|
|
|
|
# Should return a valid recordset
|
|
self.assertIsInstance(
|
|
dependents, self.env["cx.tower.jet.template"].__class__
|
|
)
|
|
|
|
# Should not include the template itself
|
|
self.assertNotIn(
|
|
template, dependents, "Template should not depend on itself"
|
|
)
|
|
|
|
def test_create_jet_with_server_logs(self):
|
|
"""Test create_jet creates server logs correctly"""
|
|
# Create a file template for server logs
|
|
file_template = self.FileTemplate.create(
|
|
{
|
|
"name": "Test Log File Template",
|
|
"file_name": "test_log.txt",
|
|
"source": "tower",
|
|
"server_dir": "/var/log",
|
|
"code": "Test log content",
|
|
}
|
|
)
|
|
|
|
# Create server logs on the template
|
|
server_log_file = self.ServerLog.create(
|
|
{
|
|
"name": "Test File Log",
|
|
"server_id": self.server_test_1.id,
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"log_type": "file",
|
|
"file_template_id": file_template.id,
|
|
"access_level": "1",
|
|
}
|
|
)
|
|
|
|
server_log_command = self.ServerLog.create(
|
|
{
|
|
"name": "Test Command Log",
|
|
"server_id": self.server_test_1.id,
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"log_type": "command",
|
|
"command_id": self.command_list_dir.id,
|
|
"access_level": "1",
|
|
}
|
|
)
|
|
|
|
# Ensure template is installed on server
|
|
self.jet_template_test.write({"server_ids": [(4, self.server_test_1.id)]})
|
|
|
|
# Create jet from template
|
|
jet = self.jet_template_test.create_jet(
|
|
server=self.server_test_1, name="Test Jet with Logs"
|
|
)
|
|
|
|
# Verify jet was created
|
|
self.assertTrue(jet, "Jet should be created")
|
|
self.assertEqual(jet.name, "Test Jet with Logs")
|
|
self.assertEqual(jet.server_id, self.server_test_1)
|
|
self.assertEqual(jet.jet_template_id, self.jet_template_test)
|
|
|
|
# Verify server logs were created for the jet
|
|
jet_logs = self.ServerLog.search([("jet_id", "=", jet.id)])
|
|
self.assertEqual(
|
|
len(jet_logs),
|
|
2,
|
|
"Should create 2 server logs (one file, one command)",
|
|
)
|
|
|
|
# Verify file-type log
|
|
jet_log_file = jet_logs.filtered(lambda log: log.log_type == "file")
|
|
self.assertEqual(
|
|
len(jet_log_file),
|
|
1,
|
|
"Should have exactly one file-type log",
|
|
)
|
|
jet_log_file = jet_log_file[0] # Get single record
|
|
self.assertEqual(
|
|
jet_log_file.jet_id,
|
|
jet,
|
|
"File log should be linked to the jet",
|
|
)
|
|
self.assertEqual(
|
|
jet_log_file.server_id,
|
|
self.server_test_1,
|
|
"File log should be linked to the server",
|
|
)
|
|
self.assertFalse(
|
|
jet_log_file.jet_template_id,
|
|
"File log should not be linked to template",
|
|
)
|
|
self.assertTrue(
|
|
jet_log_file.file_id,
|
|
"File log should have a file created",
|
|
)
|
|
self.assertEqual(
|
|
jet_log_file.file_template_id,
|
|
server_log_file.file_template_id,
|
|
"File log should reference the same file template as template log",
|
|
)
|
|
self.assertEqual(
|
|
jet_log_file.name,
|
|
server_log_file.name,
|
|
"File log should have the same name as template log",
|
|
)
|
|
self.assertEqual(
|
|
jet_log_file.file_id.jet_id,
|
|
jet,
|
|
"Created file should be linked to the jet",
|
|
)
|
|
self.assertEqual(
|
|
jet_log_file.file_id.server_id,
|
|
self.server_test_1,
|
|
"Created file should be linked to the server",
|
|
)
|
|
|
|
# Verify command-type log
|
|
jet_log_command = jet_logs.filtered(lambda log: log.log_type == "command")
|
|
self.assertEqual(
|
|
len(jet_log_command),
|
|
1,
|
|
"Should have exactly one command-type log",
|
|
)
|
|
jet_log_command = jet_log_command[0] # Get single record
|
|
self.assertEqual(
|
|
jet_log_command.jet_id,
|
|
jet,
|
|
"Command log should be linked to the jet",
|
|
)
|
|
self.assertEqual(
|
|
jet_log_command.server_id,
|
|
self.server_test_1,
|
|
"Command log should be linked to the server",
|
|
)
|
|
self.assertFalse(
|
|
jet_log_command.jet_template_id,
|
|
"Command log should not be linked to template",
|
|
)
|
|
self.assertFalse(
|
|
jet_log_command.file_id,
|
|
"Command log should not have a file",
|
|
)
|
|
self.assertEqual(
|
|
jet_log_command.command_id,
|
|
server_log_command.command_id,
|
|
"Command log should reference the same command as template log",
|
|
)
|
|
self.assertEqual(
|
|
jet_log_command.name,
|
|
server_log_command.name,
|
|
"Command log should have the same name as template log",
|
|
)
|
|
|
|
# Verify original template logs are unchanged
|
|
template_logs = self.ServerLog.search(
|
|
[("jet_template_id", "=", self.jet_template_test.id)]
|
|
)
|
|
self.assertIn(
|
|
server_log_file,
|
|
template_logs,
|
|
"Template file log should still exist",
|
|
)
|
|
self.assertIn(
|
|
server_log_command,
|
|
template_logs,
|
|
"Template command log should still exist",
|
|
)
|
|
self.assertFalse(
|
|
server_log_file.jet_id,
|
|
"Template file log should not be linked to any jet",
|
|
)
|
|
self.assertFalse(
|
|
server_log_command.jet_id,
|
|
"Template command log should not be linked to any jet",
|
|
)
|
|
|
|
def test_create_jet_with_multiple_file_logs(self):
|
|
"""Test create_jet creates multiple file logs correctly"""
|
|
# Create multiple file templates
|
|
file_template_1 = self.FileTemplate.create(
|
|
{
|
|
"name": "Log File Template 1",
|
|
"file_name": "log1.txt",
|
|
"source": "tower",
|
|
"server_dir": "/var/log",
|
|
"code": "Log 1 content",
|
|
}
|
|
)
|
|
|
|
file_template_2 = self.FileTemplate.create(
|
|
{
|
|
"name": "Log File Template 2",
|
|
"file_name": "log2.txt",
|
|
"source": "tower",
|
|
"server_dir": "/var/log",
|
|
"code": "Log 2 content",
|
|
}
|
|
)
|
|
|
|
# Create multiple server logs on the template
|
|
self.ServerLog.create(
|
|
{
|
|
"name": "File Log 1",
|
|
"server_id": self.server_test_1.id,
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"log_type": "file",
|
|
"file_template_id": file_template_1.id,
|
|
"access_level": "1",
|
|
}
|
|
)
|
|
|
|
self.ServerLog.create(
|
|
{
|
|
"name": "File Log 2",
|
|
"server_id": self.server_test_1.id,
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"log_type": "file",
|
|
"file_template_id": file_template_2.id,
|
|
"access_level": "2",
|
|
}
|
|
)
|
|
|
|
# Ensure template is installed on server
|
|
self.jet_template_test.write({"server_ids": [(4, self.server_test_1.id)]})
|
|
|
|
# Create jet from template
|
|
jet = self.jet_template_test.create_jet(
|
|
server=self.server_test_1, name="Test Jet Multiple Files"
|
|
)
|
|
|
|
# Verify all file logs were created
|
|
jet_logs = self.ServerLog.search([("jet_id", "=", jet.id)])
|
|
file_logs = jet_logs.filtered(lambda log: log.log_type == "file")
|
|
self.assertEqual(
|
|
len(file_logs),
|
|
2,
|
|
"Should create 2 file logs",
|
|
)
|
|
|
|
# Verify each file log has its own file
|
|
files = file_logs.mapped("file_id")
|
|
self.assertEqual(
|
|
len(files),
|
|
2,
|
|
"Should create 2 files",
|
|
)
|
|
self.assertEqual(
|
|
len(set(files.ids)),
|
|
2,
|
|
"Files should be different",
|
|
)
|
|
|
|
# Verify files are linked correctly
|
|
for log in file_logs:
|
|
self.assertTrue(log.file_id, "Each log should have a file")
|
|
self.assertEqual(
|
|
log.file_id.jet_id,
|
|
jet,
|
|
"File should be linked to the jet",
|
|
)
|
|
self.assertEqual(
|
|
log.file_id.server_id,
|
|
self.server_test_1,
|
|
"File should be linked to the server",
|
|
)
|
|
|
|
def test_create_jet_with_no_server_logs(self):
|
|
"""Test create_jet works correctly when template has no server logs"""
|
|
# Ensure template has no server logs
|
|
self.jet_template_test.server_log_ids.unlink()
|
|
|
|
# Ensure template is installed on server
|
|
self.jet_template_test.write({"server_ids": [(4, self.server_test_1.id)]})
|
|
|
|
# Create jet from template
|
|
jet = self.jet_template_test.create_jet(
|
|
server=self.server_test_1, name="Test Jet No Logs"
|
|
)
|
|
|
|
# Verify jet was created
|
|
self.assertTrue(jet, "Jet should be created")
|
|
|
|
# Verify no server logs were created
|
|
jet_logs = self.ServerLog.search([("jet_id", "=", jet.id)])
|
|
self.assertEqual(
|
|
len(jet_logs),
|
|
0,
|
|
"Should not create any server logs when template has none",
|
|
)
|
|
|
|
def test_create_jet_server_logs_fields_copied(self):
|
|
"""Test that server log fields are correctly copied from template"""
|
|
# Create a file template
|
|
file_template = self.FileTemplate.create(
|
|
{
|
|
"name": "Test Log File Template",
|
|
"file_name": "test_log.txt",
|
|
"source": "tower",
|
|
"server_dir": "/var/log",
|
|
"code": "Test log content",
|
|
}
|
|
)
|
|
|
|
# Create server log with various fields
|
|
server_log = self.ServerLog.create(
|
|
{
|
|
"name": "Test Log with Fields",
|
|
"server_id": self.server_test_1.id,
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"log_type": "file",
|
|
"file_template_id": file_template.id,
|
|
"access_level": "2",
|
|
"use_sudo": True,
|
|
"reference": "test_log_ref",
|
|
}
|
|
)
|
|
|
|
# Ensure template is installed on server
|
|
self.jet_template_test.write({"server_ids": [(4, self.server_test_1.id)]})
|
|
|
|
# Create jet from template
|
|
jet = self.jet_template_test.create_jet(
|
|
server=self.server_test_1, name="Test Jet Fields"
|
|
)
|
|
|
|
# Find the created log
|
|
jet_log = self.ServerLog.search([("jet_id", "=", jet.id)], limit=1)
|
|
|
|
# Verify fields are copied correctly
|
|
self.assertEqual(
|
|
jet_log.name,
|
|
server_log.name,
|
|
"Log name should be copied",
|
|
)
|
|
self.assertEqual(
|
|
jet_log.log_type,
|
|
server_log.log_type,
|
|
"Log type should be copied",
|
|
)
|
|
self.assertEqual(
|
|
jet_log.file_template_id,
|
|
server_log.file_template_id,
|
|
"File template should be copied",
|
|
)
|
|
self.assertEqual(
|
|
jet_log.access_level,
|
|
server_log.access_level,
|
|
"Access level should be copied",
|
|
)
|
|
self.assertEqual(
|
|
jet_log.use_sudo,
|
|
server_log.use_sudo,
|
|
"Use sudo should be copied",
|
|
)
|
|
# Reference should be different (due to reference mixin)
|
|
self.assertNotEqual(
|
|
jet_log.reference,
|
|
server_log.reference,
|
|
"Reference should be different (unique)",
|
|
)
|
|
# Verify file was created for file-type log
|
|
self.assertTrue(
|
|
jet_log.file_id,
|
|
"File should be created for file-type log",
|
|
)
|
|
self.assertEqual(
|
|
jet_log.file_id.jet_id,
|
|
jet,
|
|
"Created file should be linked to the jet",
|
|
)
|
|
|
|
def test_create_jet_different_servers(self):
|
|
"""Test create_jet creates logs with correct server_id for different servers"""
|
|
# Create a file template
|
|
file_template = self.FileTemplate.create(
|
|
{
|
|
"name": "Test Log File Template",
|
|
"file_name": "test_log.txt",
|
|
"source": "tower",
|
|
"server_dir": "/var/log",
|
|
"code": "Test log content",
|
|
}
|
|
)
|
|
|
|
# Create server log on template (linked to server_test_1)
|
|
self.ServerLog.create(
|
|
{
|
|
"name": "Test Log",
|
|
"server_id": self.server_test_1.id,
|
|
"jet_template_id": self.jet_template_test.id,
|
|
"log_type": "file",
|
|
"file_template_id": file_template.id,
|
|
}
|
|
)
|
|
|
|
# Ensure template is installed on both servers
|
|
self.jet_template_test.write(
|
|
{
|
|
"server_ids": [
|
|
(4, self.server_test_1.id),
|
|
(4, self.server_test_2.id),
|
|
]
|
|
}
|
|
)
|
|
|
|
# Create jet on server_test_2
|
|
jet = self.jet_template_test.create_jet(
|
|
server=self.server_test_2, name="Test Jet Server 2"
|
|
)
|
|
|
|
# Verify jet was created on correct server
|
|
self.assertEqual(
|
|
jet.server_id,
|
|
self.server_test_2,
|
|
"Jet should be on server_test_2",
|
|
)
|
|
|
|
# Verify server log is linked to server_test_2
|
|
jet_log = self.ServerLog.search([("jet_id", "=", jet.id)], limit=1)
|
|
self.assertEqual(
|
|
jet_log.server_id,
|
|
self.server_test_2,
|
|
"Server log should be linked to server_test_2",
|
|
)
|
|
self.assertEqual(
|
|
jet_log.file_id.server_id,
|
|
self.server_test_2,
|
|
"File should be linked to server_test_2",
|
|
)
|