odoo-tower/odoo-addons
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Wipe cetmix_tower_yaml (polluted by overlapping uploads)

Browse Source
This commit is contained in:
Tower Deploy
2026-04-27 13:43:58 +03:00
parent 18dd9c7a1f
commit 7cef9f1a32
80 changed files with 0 additions and 9275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
addons/cetmix_tower_yaml/tests/__init__.py
View File

@@ -1,8 +0,0 @@
from . import test_command
from . import test_tower_yaml_mixin
from . import test_file_template
from . import test_plan
from . import test_yaml_export_wizard
from . import test_yaml_import_wizard
from . import test_server_log
from . import test_server_yaml

347
addons/cetmix_tower_yaml/tests/test_command.py
View File

@@ -1,347 +0,0 @@
# Copyright (C) 2024 Cetmix OÜ
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import yaml
from odoo.exceptions import ValidationError
from odoo.tests import TransactionCase
class TestTowerCommand(TransactionCase):
@classmethod
def setUpClass(cls, *args, **kwargs):
super().setUpClass(*args, **kwargs)
cls.Command = cls.env["cx.tower.command"]
# Expected YAML content of the test command
cls.command_test_yaml = """cetmix_tower_model: command
access_level: manager
reference: test_yaml_in_tests
name: Test YAML
action: ssh_command
allow_parallel_run: false
note: |-
Test YAML command conversion.
Ensure all fields are rendered properly.
os_ids: false
tag_ids: false
path: false
file_template_id: false
flight_plan_id: false
code: |-
cd /home/{{ tower.server.ssh_username }} \\
&& ls -lha
server_status: false
variable_ids: false
secret_ids: false
no_split_for_sudo: false
if_file_exists: skip
disconnect_file: false
"""
# YAML content translated into Python dict
cls.command_test_yaml_dict = yaml.safe_load(cls.command_test_yaml)
def test_yaml_from_command(self):
"""Test if YAML is generated properly from a command"""
# -- 0 --
# Create test command
# Test command
command_test = self.Command.create(
{
"name": "Test YAML",
"reference": "test_yaml_in_tests",
"action": "ssh_command",
"code": """cd /home/{{ tower.server.ssh_username }} \\
&& ls -lha""",
"note": """Test YAML command conversion.
Ensure all fields are rendered properly.""",
}
)
# -- 1 --
# Check it YAML generated by the command matches
# YAML from the template file
self.assertEqual(
command_test.yaml_code,
self.command_test_yaml,
"YAML generated from command doesn't match template file one",
)
# -- 2 --
# Check if YAML key values match Cetmix Tower ones
self.assertEqual(
command_test.access_level,
self.Command.TO_TOWER_ACCESS_LEVEL[
self.command_test_yaml_dict["access_level"]
],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.action,
self.command_test_yaml_dict["action"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.allow_parallel_run,
self.command_test_yaml_dict["allow_parallel_run"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.code,
self.command_test_yaml_dict["code"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.name,
self.command_test_yaml_dict["name"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.note,
self.command_test_yaml_dict["note"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.path,
self.command_test_yaml_dict["path"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.reference,
self.command_test_yaml_dict["reference"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.if_file_exists,
self.command_test_yaml_dict["if_file_exists"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command_test.disconnect_file,
self.command_test_yaml_dict["disconnect_file"],
"YAML value doesn't match Cetmix Tower one",
)
def test_command_from_yaml(self):
"""Test if YAML is generated properly from a command"""
def test_yaml(command):
"""Checks if yaml values are inserted correctly
Args:
command(cx.tower.command): _description_
"""
self.assertEqual(
command.access_level,
self.Command.TO_TOWER_ACCESS_LEVEL[
self.command_test_yaml_dict["access_level"]
],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.action,
self.command_test_yaml_dict["action"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.allow_parallel_run,
self.command_test_yaml_dict["allow_parallel_run"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.code,
self.command_test_yaml_dict["code"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.name,
self.command_test_yaml_dict["name"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.note,
self.command_test_yaml_dict["note"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.path,
self.command_test_yaml_dict["path"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.reference,
self.command_test_yaml_dict["reference"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.if_file_exists,
self.command_test_yaml_dict["if_file_exists"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
command.disconnect_file,
self.command_test_yaml_dict["disconnect_file"],
"YAML value doesn't match Cetmix Tower one",
)
# Create test command
command_test = self.Command.create(
{"name": "New Command", "action": "python_code"}
)
# -- 1 --
# Insert YAML into the command and
# check if YAML key values match Cetmix Tower ones
command_test.yaml_code = self.command_test_yaml
test_yaml(command_test)
# -- 2 --
# Insert some non supported keys and ensure nothing bad happens
yaml_with_non_supported_keys = """access_level: manager
action: ssh_command
doge: wow
memes: much nice!
allow_parallel_run: false
cetmix_tower_model: command
code: |-
cd /home/{{ tower.server.ssh_username }} \\
&& ls -lha
file_template_id: false
flight_plan_id: false
name: Test YAML
note: |-
Test YAML command conversion.
Ensure all fields are rendered properly.
path: false
reference: test_yaml_in_tests
tag_ids: false
"""
command_test.yaml_code = yaml_with_non_supported_keys
test_yaml(command_test)
# -- 3 --
# Insert non existing selection field value and exception is raised
yaml_with_non_supported_keys = """access_level: manager
action: non_existing_action
doge: wow
memes: much nice!
allow_parallel_run: false
cetmix_tower_model: command
code: |-
cd /home/{{ tower.server.ssh_username }} \\
&& ls -lha
file_template_id: false
flight_plan_id: false
name: Test YAML
note: |-
Test YAML command conversion.
Ensure all fields are rendered properly.
path: false
reference: test_yaml_in_tests
tag_ids: false
"""
with self.assertRaises(ValidationError) as e:
command_test.yaml_code = yaml_with_non_supported_keys
self.assertIn("non_existing_action", str(e.exception))
self.assertEqual(
str(e),
"Wrong value for cx.tower.command.action: 'non_existing_action'",
"Exception message doesn't match",
)
def test_command_with_action_file_template(self):
"""Test command with 'File from template' action"""
yaml_with_reference = """cetmix_tower_model: command
access_level: manager
reference: such_much_test_command
name: Such Much Command
action: file_using_template
allow_parallel_run: false
note: Just a note
os_ids: false
tag_ids: false
path: false
file_template_id: my_custom_test_template
flight_plan_id: false
code: false
server_status: false
variable_ids: false
secret_ids: false
no_split_for_sudo: false
if_file_exists: skip
disconnect_file: false
"""
# Add file template
file_template = self.env["cx.tower.file.template"].create(
{
"name": "Such much demo",
"reference": "my_custom_test_template",
"file_name": "much_logs.txt",
"server_dir": "/var/log/my/files",
"source": "tower",
"file_type": "text",
"note": "Hey!",
"keep_when_deleted": False,
}
)
command_with_template = self.Command.create(
{
"name": "Such Much Command",
"reference": "such_much_test_command",
"action": "file_using_template",
"note": "Just a note",
"file_template_id": file_template.id,
}
)
# -- 1 --
# Check if final YAML composed correctly
self.assertEqual(
command_with_template.yaml_code,
yaml_with_reference,
"YAML is not composed correctly",
)
# -- 2 --
# Explode related record and check the YAML
yaml_with_reference_exploded = """cetmix_tower_model: command
access_level: manager
reference: such_much_test_command
name: Such Much Command
action: file_using_template
allow_parallel_run: false
note: Just a note
os_ids: false
tag_ids: false
path: false
file_template_id:
reference: my_custom_test_template
name: Such much demo
source: tower
file_type: text
server_dir: /var/log/my/files
file_name: much_logs.txt
keep_when_deleted: false
tag_ids: false
note: Hey!
code: false
variable_ids: false
secret_ids: false
flight_plan_id: false
code: false
server_status: false
variable_ids: false
secret_ids: false
no_split_for_sudo: false
if_file_exists: skip
disconnect_file: false
"""
command_with_template.invalidate_recordset(["yaml_code"])
self.assertEqual(
command_with_template.with_context(explode_related_record=True).yaml_code,
yaml_with_reference_exploded,
"YAML is not composed correctly",
)

320
addons/cetmix_tower_yaml/tests/test_file_template.py
View File

@@ -1,320 +0,0 @@
import yaml
from odoo.tests import TransactionCase
class TestTowerFileTemplate(TransactionCase):
@classmethod
def setUpClass(cls, *args, **kwargs):
super().setUpClass(*args, **kwargs)
cls.FileTemplate = cls.env["cx.tower.file.template"]
# Expected YAML content of the test file template
cls.file_template_test_yaml = """cetmix_tower_model: file_template
reference: dockerfile_unit_test
name: Dockerfile Test
source: tower
file_type: text
server_dir: /opt
file_name: Dockerfile
keep_when_deleted: true
tag_ids: false
note: |-
Used to build Odoo addons image.
Depends on Odoo core image.
code: |-
FROM odoo:{{ odoo_test_version }}
# Install git-aggregator and tools for requirements generation
RUN pip3 install --upgrade pip && pip install manifestoo setuptools-odoo git-aggregator
# Let's go!
USER odoo
variable_ids: false
secret_ids: false
""" # noqa
# Expected YAML content of the test file template
# without empty x2mvalues
cls.file_template_test_yaml_no_empty_values = """cetmix_tower_model: file_template
reference: dockerfile_unit_test
name: Dockerfile Test
source: tower
file_type: text
server_dir: /opt
file_name: Dockerfile
keep_when_deleted: true
note: |-
Used to build Odoo addons image.
Depends on Odoo core image.
code: |-
FROM odoo:{{ odoo_test_version }}
# Install git-aggregator and tools for requirements generation
RUN pip3 install --upgrade pip && pip install manifestoo setuptools-odoo git-aggregator
# Let's go!
USER odoo
""" # noqa
# YAML content translated into Python dict
cls.file_template_test_yaml_dict = yaml.safe_load(cls.file_template_test_yaml)
cls.file_template_test_yaml_dict_no_empty_values = yaml.safe_load(
cls.file_template_test_yaml_no_empty_values
)
def test_yaml_from_file_template(self):
"""Test if YAML is generated properly from a file"""
# -- 0 --
# Create test file
# Test file
file_template_test = self.FileTemplate.create(
{
"name": "Dockerfile Test",
"reference": "dockerfile_unit_test",
"file_name": "Dockerfile",
"server_dir": "/opt",
"source": "tower",
"keep_when_deleted": True,
"file_type": "text",
"code": """FROM odoo:{{ odoo_test_version }}
# Install git-aggregator and tools for requirements generation
RUN pip3 install --upgrade pip && pip install manifestoo setuptools-odoo git-aggregator
# Let's go!
USER odoo""",
"note": """Used to build Odoo addons image.
Depends on Odoo core image.""",
}
)
# -- 1 --
# Check it YAML generated by the file matches
# YAML from the template file
self.assertEqual(
file_template_test.yaml_code,
self.file_template_test_yaml,
"YAML generated from file doesn't match template file one",
)
# -- 2 --
# Check if YAML key values match Cetmix Tower ones
self.assertEqual(
file_template_test.source,
self.file_template_test_yaml_dict["source"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.file_name,
self.file_template_test_yaml_dict["file_name"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.code,
self.file_template_test_yaml_dict["code"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.name,
self.file_template_test_yaml_dict["name"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.note,
self.file_template_test_yaml_dict["note"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.server_dir,
self.file_template_test_yaml_dict["server_dir"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.reference,
self.file_template_test_yaml_dict["reference"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.file_type,
self.file_template_test_yaml_dict["file_type"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.keep_when_deleted,
self.file_template_test_yaml_dict["keep_when_deleted"],
"YAML value doesn't match Cetmix Tower one",
)
def test_yaml_from_file_template_no_empty_values(self):
"""Test if YAML is generated properly from a file"""
# -- 0 --
# Create test file
# Test file
file_template_test = self.FileTemplate.with_context(
remove_empty_values=True
).create(
{
"name": "Dockerfile Test",
"reference": "dockerfile_unit_test",
"file_name": "Dockerfile",
"server_dir": "/opt",
"source": "tower",
"keep_when_deleted": True,
"file_type": "text",
"code": """FROM odoo:{{ odoo_test_version }}
# Install git-aggregator and tools for requirements generation
RUN pip3 install --upgrade pip && pip install manifestoo setuptools-odoo git-aggregator
# Let's go!
USER odoo""",
"note": """Used to build Odoo addons image.
Depends on Odoo core image.""",
}
)
# -- 1 --
# Check it YAML generated by the file matches
# YAML from the template file
self.assertEqual(
file_template_test.yaml_code,
self.file_template_test_yaml_no_empty_values,
"YAML generated from file doesn't match template file one",
)
# -- 2 --
# Check if YAML key values match Cetmix Tower ones
self.assertEqual(
file_template_test.source,
self.file_template_test_yaml_dict_no_empty_values["source"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.file_name,
self.file_template_test_yaml_dict_no_empty_values["file_name"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.code,
self.file_template_test_yaml_dict_no_empty_values["code"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.name,
self.file_template_test_yaml_dict_no_empty_values["name"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.note,
self.file_template_test_yaml_dict_no_empty_values["note"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.server_dir,
self.file_template_test_yaml_dict_no_empty_values["server_dir"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.reference,
self.file_template_test_yaml_dict_no_empty_values["reference"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.file_type,
self.file_template_test_yaml_dict_no_empty_values["file_type"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template_test.keep_when_deleted,
self.file_template_test_yaml_dict_no_empty_values["keep_when_deleted"],
"YAML value doesn't match Cetmix Tower one",
)
def test_file_template_from_yaml(self):
"""Test if YAML is generated properly from a file"""
def test_yaml(file_template):
"""Checks if yaml values are inserted correctly
Args:
file_template (cx.tower.file.template): File template
"""
self.assertEqual(
file_template.source,
self.file_template_test_yaml_dict["source"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template.file_name,
self.file_template_test_yaml_dict["file_name"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template.code,
self.file_template_test_yaml_dict["code"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template.name,
self.file_template_test_yaml_dict["name"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template.note,
self.file_template_test_yaml_dict["note"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template.server_dir,
self.file_template_test_yaml_dict["server_dir"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template.reference,
self.file_template_test_yaml_dict["reference"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template.file_type,
self.file_template_test_yaml_dict["file_type"],
"YAML value doesn't match Cetmix Tower one",
)
self.assertEqual(
file_template.keep_when_deleted,
self.file_template_test_yaml_dict["keep_when_deleted"],
"YAML value doesn't match Cetmix Tower one",
)
# Create test file template
file_template_test = self.FileTemplate.create({"name": "New file template"})
# -- 1 --
# Insert YAML into the file and
# check if YAML key values match Cetmix Tower ones
file_template_test.yaml_code = self.file_template_test_yaml
test_yaml(file_template_test)
# -- 2 --
# Insert some non supported keys and ensure nothing bad happens
yaml_with_non_supported_keys = """cetmix_tower_model: file_template
code: |-
FROM odoo:{{ odoo_test_version }}
# Install git-aggregator and tools for requirements generation
RUN pip3 install --upgrade pip && pip install manifestoo setuptools-odoo git-aggregator
# Let's go!
USER odoo
doge: SoMuch style!
file_name: Dockerfile
file_type: text
keep_when_deleted: true
name: Dockerfile Test
note: |-
Used to build Odoo addons image.
Depends on Odoo core image.
reference: dockerfile_unit_test
server_dir: /opt
source: tower
tag_ids: false
""" # noqa
file_template_test.yaml_code = yaml_with_non_supported_keys
test_yaml(file_template_test)

179
addons/cetmix_tower_yaml/tests/test_plan.py
View File

@@ -1,179 +0,0 @@
from odoo.tests import TransactionCase
class TestTowerPlan(TransactionCase):
@classmethod
def setUpClass(cls, *args, **kwargs):
super().setUpClass(*args, **kwargs)
cls.Plan = cls.env["cx.tower.plan"]
def test_plan_create_from_yaml(self):
"""Test plan creation from YAML."""
plan_yaml = """cetmix_tower_model: plan
access_level: manager
reference: test_plan_from_yaml
name: 'Test Plan From Yaml'
allow_parallel_run: false
color: 0
tag_ids:
- reference: doge_test_plan_tag
name: Doge Test Plan Tag
color: 1
on_error_action: e
custom_exit_code: 0
line_ids:
- sequence: 5
condition: false
use_sudo: false
path: /such/much/{{ test_plan_dir }}
command_id:
access_level: manager
reference: very_much_command_test
name: Very much command
action: ssh_command
allow_parallel_run: false
note: false
code: Such much code
variable_ids:
- cetmix_tower_model: variable
reference: test_plan_dir
name: Test Plan Directory
action_ids:
- sequence: 1
condition: ==
value_char: '0'
action: n
custom_exit_code: 0
variable_value_ids:
- cetmix_tower_model: variable_value
variable_id:
cetmix_tower_yaml_version: 1
cetmix_tower_model: variable
reference: test_plan_branch
name: Test Plan Branch
value_char: production
- cetmix_tower_model: variable_value
variable_id:
cetmix_tower_yaml_version: 1
cetmix_tower_model: variable
reference: test_plan_some_unique_variable
name: Test Plan Some Unique Variable
value_char: 'Final Value'
- cetmix_tower_model: plan_line_action
access_level: manager
sequence: 2
condition: '>'
value_char: '0'
action: ec
custom_exit_code: 255
variable_value_ids: false
variable_ids: false
"""
# -- 1 --
# Create plan from YAML
plan_form_yaml = self.Plan.create(
{"name": "Name Placeholder", "yaml_code": plan_yaml}
)
self.assertEqual(
plan_form_yaml.reference,
"test_plan_from_yaml",
"Reference is not set from YAML",
)
# Name should be set from YAML
self.assertEqual(
plan_form_yaml.name, "Test Plan From Yaml", "Name is not set from YAML"
)
# -- 2 --
# Check plan tags
plan_tags = plan_form_yaml.tag_ids
self.assertEqual(len(plan_tags), 1)
self.assertEqual(plan_tags.name, "Doge Test Plan Tag")
# -- 3 --
# Check plan lines
plan_lines = plan_form_yaml.line_ids
self.assertEqual(len(plan_lines), 1, "Line count is not 1")
self.assertFalse(plan_lines.condition, "Condition is not false")
self.assertEqual(
plan_lines.path,
"/such/much/{{ test_plan_dir }}",
"Path is not set from YAML",
)
self.assertEqual(
plan_lines.command_id.reference,
"very_much_command_test",
"Command reference is not set from YAML",
)
self.assertEqual(
plan_lines.command_id.name,
"Very much command",
"Command name is not set from YAML",
)
self.assertEqual(
plan_lines.command_id.action,
"ssh_command",
"Command action is not set from YAML",
)
self.assertFalse(
plan_lines.command_id.allow_parallel_run,
"Command allow parallel run is not set from YAML",
)
self.assertFalse(
plan_lines.command_id.note, "Command note is not set from YAML"
)
self.assertEqual(
plan_lines.command_id.variable_ids.mapped("reference"),
["test_plan_dir"],
"Command variable ids is not set from YAML",
)
self.assertEqual(
plan_lines.command_id.access_level,
"2",
"Command access level is not set from YAML",
)
# -- 4 --
# Check plan line actions
plan_actions = plan_form_yaml.line_ids.action_ids
self.assertEqual(len(plan_actions), 2, "Action count is not 2")
self.assertEqual(
plan_actions[0].condition, "==", "First action condition is not equal"
)
self.assertEqual(
plan_actions[0].value_char, "0", "First action value char is not 0"
)
self.assertEqual(plan_actions[0].action, "n", "First action action is not n")
self.assertEqual(
plan_actions[0].custom_exit_code,
0,
"First action custom exit code is not 0",
)
self.assertEqual(
len(plan_actions[0].variable_value_ids),
2,
"Number of variable value ids is not correct",
)
self.assertEqual(
plan_actions[0].variable_value_ids.mapped("value_char"),
["production", "Final Value"],
"Variable value chars are not correct",
)
self.assertEqual(
plan_actions[1].condition, ">", "Second action condition is not greater"
)
self.assertEqual(
plan_actions[1].value_char, "0", "Second action value char is not 0"
)
self.assertEqual(plan_actions[1].action, "ec", "Second action action is not ec")
self.assertEqual(
plan_actions[1].custom_exit_code,
255,
"Second action custom exit code is not 255",
)
self.assertFalse(
plan_actions[1].variable_value_ids,
"Second action variable value ids is not false",
)

127
addons/cetmix_tower_yaml/tests/test_server_log.py
View File

@@ -1,127 +0,0 @@
# Copyright (C) 2025 Cetmix OÜ
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
"""
Tests for the cx.tower.server.log model YAML export/import.
Covers:
1. YAML export of a file-type log must include `file_id` and allow suffixes.
2. A full round-trip (export → delete → import) preserves the `file_id` relation.
3. Exporting a non-file log must include a falsy `file_id`.
4. Importing YAML with a bogus `file_id` reference raises ValidationError.
"""
import yaml
from odoo.tests import TransactionCase, tagged
@tagged("post_install", "-at_install")
class TestServerLog(TransactionCase):
"""YAML export/import tests for cx.tower.server.log."""
@classmethod
def setUpClass(cls):
super().setUpClass()
env = cls.env
cls.File = env["cx.tower.file"]
cls.Server = env["cx.tower.server"]
cls.ServerLog = env["cx.tower.server.log"]
# Create a file to reference from the log
cls.file = cls.File.create(
{
"name": "repos.yaml",
"reference": "reposyaml",
"source": "tower",
"file_type": "text",
"server_dir": "/tmp",
"code": "# Example\nHello, Tower!",
}
)
# Create a server (use password auth to satisfy constraints)
cls.server = cls.Server.create(
{
"name": "Srv-YAML-Test",
"reference": "srv_yaml_test",
"ip_v4_address": "127.0.0.1",
"ssh_username": "admin",
"ssh_port": 22,
"ssh_auth_mode": "p",
"ssh_password": "dummy",
"use_sudo": False,
}
)
# Create a file-type log linked to the file above
cls.log = cls.ServerLog.create(
{
"name": "Log from file",
"reference": "log_from_file",
"log_type": "file",
"file_id": cls.file.id,
"server_id": cls.server.id,
"use_sudo": False,
}
)
def test_yaml_export_contains_file_id(self):
"""Exported YAML must include a file_id starting with the file's reference."""
data = yaml.safe_load(self.log.yaml_code)
# Ensure file_id is present
self.assertIn("file_id", data, "`file_id` is missing from YAML export")
# Allow for auto-appended suffixes, so only check prefix
self.assertTrue(
data["file_id"].startswith(self.file.reference),
f"`file_id` value '{data['file_id']}' should start with "
f"'{self.file.reference}'",
)
def test_yaml_roundtrip_restores_file_id(self):
"""A full export→delete→import cycle must restore the file_id relation."""
yaml_dict = yaml.safe_load(self.log.yaml_code)
# Remove the original log
self.log.unlink()
# Recreate from YAML
vals = self.ServerLog._post_process_yaml_dict_values(yaml_dict)
restored = self.ServerLog.with_context(from_yaml=True).create(vals)
# Verify relation restored
self.assertEqual(
restored.file_id.id,
self.file.id,
"`file_id` was not restored after round-trip",
)
def test_yaml_export_without_file_id(self):
"""Logs of non-file type should not include file_id in YAML."""
cmd_log = self.ServerLog.create(
{
"name": "Log no file",
"reference": "log_no_file",
"log_type": "command",
"server_id": self.server.id,
"use_sudo": False,
}
)
data = yaml.safe_load(cmd_log.yaml_code)
# key is present, but must be falsy
self.assertIn("file_id", data, "`file_id` key is missing")
self.assertFalse(
data["file_id"],
"`file_id` for non-file log must be False/empty",
)
def test_yaml_import_with_missing_file_reference(self):
"""Missing file reference is accepted, but file_id stays empty."""
yaml_dict = yaml.safe_load(self.log.yaml_code)
yaml_dict["file_id"] = "does_not_exist"
vals = self.ServerLog._post_process_yaml_dict_values(yaml_dict)
new_log = self.ServerLog.with_context(from_yaml=True).create(vals)
# Log is created, but the relation is not resolved
self.assertFalse(
new_log.file_id,
"file_id should be empty when reference cannot be resolved",
)

124
addons/cetmix_tower_yaml/tests/test_server_yaml.py
View File

@@ -1,124 +0,0 @@
# Copyright (C) 2025 Cetmix OÜ
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
"""
Tests for cx.tower.server YAML export/import covering command_ids and plan_ids.
"""
import yaml
from odoo.tests import TransactionCase, tagged
@tagged("post_install", "-at_install")
class TestServerYAML(TransactionCase):
"""YAML export/import tests for cx.tower.server with commands and plans."""
@classmethod
def setUpClass(cls):
super().setUpClass()
env = cls.env
cls.Server = env["cx.tower.server"]
cls.Command = env["cx.tower.command"]
cls.Plan = env["cx.tower.plan"]
# Create a command to attach (use defaults for access_level)
cls.command = cls.Command.create(
{
"name": "Test Command",
"reference": "test_command",
"action": "ssh_command",
"allow_parallel_run": False,
}
)
# Create a flight plan to attach
cls.plan = cls.Plan.create(
{
"name": "Test Flight Plan",
"reference": "test_plan",
"allow_parallel_run": False,
"color": 0,
}
)
# Create server and link command and plan
cls.server = cls.Server.create(
{
"name": "Server YAML Test",
"reference": "srv_yaml_test",
"ip_v4_address": "127.0.0.1",
"ssh_username": "admin",
"ssh_port": 22,
"ssh_auth_mode": "p",
"ssh_password": "dummy",
"use_sudo": False,
# Link the m2m fields
"command_ids": [(6, 0, [cls.command.id])],
"plan_ids": [(6, 0, [cls.plan.id])],
}
)
def test_yaml_export_contains_command_and_plan(self):
"""Exported YAML include command_ids and plan_ids with correct references."""
data = yaml.safe_load(self.server.yaml_code)
# Check command_ids
self.assertIn(
"command_ids",
data,
"`command_ids` is missing from YAML export",
)
self.assertIsInstance(
data["command_ids"], list, "`command_ids` should be a list in YAML"
)
self.assertTrue(
data["command_ids"],
"`command_ids` list should not be empty",
)
# Only reference should be exported
self.assertEqual(
data["command_ids"][0],
self.command.reference,
"Exported command reference does not match",
)
# Check plan_ids
self.assertIn(
"plan_ids",
data,
"`plan_ids` is missing from YAML export",
)
self.assertIsInstance(
data["plan_ids"], list, "`plan_ids` should be a list in YAML"
)
self.assertTrue(
data["plan_ids"],
"`plan_ids` list should not be empty",
)
self.assertEqual(
data["plan_ids"][0],
self.plan.reference,
"Exported plan reference does not match",
)
def test_yaml_roundtrip_restores_command_and_plan(self):
"""A full export→delete→import cycle must restore the m2m relations."""
yaml_dict = yaml.safe_load(self.server.yaml_code)
# Remove original server
self.server.unlink()
# Prepare values and import
vals = self.Server._post_process_yaml_dict_values(yaml_dict)
restored = self.Server.with_context(
from_yaml=True, skip_ssh_settings_check=True
).create(vals)
# Verify m2m links restored
self.assertEqual(
restored.command_ids.ids,
[self.command.id],
"`command_ids` were not restored correctly",
)
self.assertEqual(
restored.plan_ids.ids,
[self.plan.id],
"`plan_ids` were not restored correctly",
)

525
addons/cetmix_tower_yaml/tests/test_tower_yaml_mixin.py
View File

@@ -1,525 +0,0 @@
# Copyright (C) 2024 Cetmix OÜ
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
from odoo import _
from odoo.exceptions import AccessError, ValidationError
from odoo.tests import TransactionCase, tagged
class TestTowerYamlMixin(TransactionCase):
@classmethod
def setUpClass(cls, *args, **kwargs):
super().setUpClass(*args, **kwargs)
cls.Users = cls.env["res.users"].with_context(no_reset_password=True)
cls.YamlMixin = cls.env["cx.tower.yaml.mixin"]
TowerTag = cls.env["cx.tower.tag"]
cls.tag_doge = TowerTag.create({"name": "Doge", "reference": "doge"})
cls.tag_pepe = TowerTag.create({"name": "Pepe", "reference": "pepe"})
def test_convert_dict_to_yaml(self):
# -- 1 --
# Test regular flow
self.assertEqual(
self.YamlMixin._convert_dict_to_yaml({"a": 1, "b": 2}),
"a: 1\nb: 2\n",
"Dictionary was not converted to YAML correctly",
)
# -- 2 --
# Test flow with exception due to wrong values
with self.assertRaises(ValidationError) as e:
self.YamlMixin._convert_dict_to_yaml("not_a_dict")
self.assertEqual(
str(e.exception),
_("Values must be a dictionary"),
"Exception message doesn't match",
)
def test_yaml_field_access(self):
# Create Root user with no access to the 'yaml_code field
user_root = self.Users.create(
{
"name": "Root User",
"login": "root@example.com",
"groups_id": [
(4, self.env.ref("base.group_user").id),
(4, self.env.ref("cetmix_tower_server.group_root").id),
],
}
)
with self.assertRaises(AccessError):
self.tag_doge.with_user(user_root).read(["yaml_code"])
# Add user to the 'cetmix_tower_yaml.group_export' group
# and check if access is granted
user_root.write(
{"groups_id": [(4, self.env.ref("cetmix_tower_yaml.group_export").id)]}
)
yaml_code = (
self.tag_doge.with_user(user_root).read(["yaml_code"])[0].get("yaml_code")
)
# Modify YAML code and check if it's saved
yaml_code = yaml_code.replace("Doge", "WowDoge")
with self.assertRaises(AccessError):
self.tag_doge.with_user(user_root).write({"yaml_code": yaml_code})
# Add user to the 'cetmix_tower_yaml.group_import' group
# and check if access is granted
user_root.write(
{"groups_id": [(4, self.env.ref("cetmix_tower_yaml.group_import").id)]}
)
self.tag_doge.with_user(user_root).write({"yaml_code": yaml_code})
self.assertEqual(
self.tag_doge.with_user(user_root).yaml_code,
yaml_code,
"YAML code was not saved",
)
def test_post_process_record_values(self):
"""Test value post processing.
We test common fields only because this method can be overridden
in models inheriting this mixin.
"""
# Patch method to return "access_level" field too
def _get_fields_for_yaml(self):
return ["access_level", "name", "reference"]
self.YamlMixin._patch_method("_get_fields_for_yaml", _get_fields_for_yaml)
source_values = {
"access_level": "3",
"id": 22332,
"name": "Doge Much Like",
"reference": "such_much_doge",
}
result_values = self.YamlMixin._post_process_record_values(source_values.copy())
self.assertNotIn("id", result_values, "ID must be removed")
self.assertEqual(
result_values["access_level"],
self.YamlMixin.TO_YAML_ACCESS_LEVEL[source_values["access_level"]],
"Access level is not parsed correctly",
)
self.assertEqual(
result_values["name"],
source_values["name"],
"Other values should remain unchanged",
)
self.assertEqual(
result_values["reference"],
source_values["reference"],
"Other values should remain unchanged",
)
# Restore original method
self.YamlMixin._revert_method("_get_fields_for_yaml")
def test_post_process_yaml_dict_values(self):
"""Test YAML dict value post processing.
We test common fields only because this method can be overridden
in models inheriting this mixin.
"""
# Patch method to return "access_level" field too
def _get_fields_for_yaml(self):
return ["access_level", "name", "reference"]
self.YamlMixin._patch_method("_get_fields_for_yaml", _get_fields_for_yaml)
# -- 1 --
# Test regular flow
source_values = {
"access_level": "user",
"name": "Doge Much Like",
"reference": "such_much_doge",
"some_doge_field": "some_meme",
}
result_values = self.YamlMixin._post_process_yaml_dict_values(
source_values.copy()
)
self.assertNotIn(
"some_doge_field", result_values, "Non listed fields must be removed"
)
self.assertEqual(
result_values["access_level"],
self.YamlMixin.TO_TOWER_ACCESS_LEVEL[source_values["access_level"]],
"Access level is not parsed correctly",
)
self.assertEqual(
result_values["name"],
source_values["name"],
"Other values should remain unchanged",
)
self.assertEqual(
result_values["reference"],
source_values["reference"],
"Other values should remain unchanged",
)
# -- Test 2 --
# Submit wrong value for access level
source_values.update(
{
"access_level": "doge",
}
)
with self.assertRaises(ValidationError) as e:
result_values = self.YamlMixin._post_process_yaml_dict_values(
source_values.copy()
)
self.assertEqual(
str(e.exception),
_(
"Wrong value for 'access_level' key: %(acv)s",
acv="doge",
),
"Exception message doesn't match",
)
# Restore original method
self.YamlMixin._revert_method("_get_fields_for_yaml")
def test_process_relation_field_value_no_explode(self):
"""Test non exploded related field values.
Non exploded values represent related record with reference only.
Covers the following child functions:
- _process_m2o_value(..)
- _process_x2m_values(..)
"""
# We are using command with file template for that
file_template = self.env["cx.tower.file.template"].create(
{"name": "Test m2o", "reference": "test_m2o"}
)
command = self.env["cx.tower.command"].create(
{
"name": "Command test m2o",
"action": "file_using_template",
"file_template_id": file_template.id,
"tag_ids": [(4, self.tag_doge.id), (4, self.tag_pepe.id)],
}
)
# -- 1 --
# Record -> Yaml
# -- 1.1 --
# Many2one
result = command._process_relation_field_value(
field="file_template_id",
value=(command.file_template_id.id, command.file_template_id.name),
record_mode=True,
)
self.assertEqual(
result, file_template.reference, "Reference was not resolved correctly"
)
# -- 1.2 --
# Many2many
result = command._process_relation_field_value(
field="tag_ids",
value=[self.tag_doge.id, self.tag_pepe.id],
record_mode=True,
)
self.assertEqual(len(result), 2, "Must be 2 references")
self.assertIn(
self.tag_doge.reference, result, "Reference was not resolved correctly"
)
self.assertIn(
self.tag_pepe.reference, result, "Reference was not resolved correctly"
)
# -- 2 --
# Yaml -> Record
# -- 2.1. --
# Many2one
result = command._process_relation_field_value(
field="file_template_id", value=file_template.reference, record_mode=False
)
self.assertEqual(
result, file_template.id, "Record ID was not resolved correctly"
)
# -- 2.2 --
# Many2many
result = command._process_relation_field_value(
field="tag_ids",
value=[self.tag_doge.reference, self.tag_pepe.reference],
record_mode=False,
)
self.assertEqual(len(result), 2, "Must be 2 records")
self.assertIn(
(4, self.tag_doge.id), result, "Record ID was not resolved correctly"
)
self.assertIn(
(4, self.tag_pepe.id), result, "Record ID was not resolved correctly"
)
# -- 3 --
# Yaml with non existing reference -> Record
result = command._process_relation_field_value(
field="file_template_id", value="such_much_not_reference", record_mode=False
)
self.assertFalse(result, "Must be 'False'")
# -- 4 --
# No record -> Yaml
result = command._process_relation_field_value(
field="file_template_id",
value=self.env["cx.tower.file.template"],
record_mode=True,
)
self.assertFalse(result, "Result must be 'False'")
def test_process_relation_field_value_explode(self):
"""Test exploded related field values.
Exploded values represent related record with a child YAML structure.
Covers the following child functions:
- _process_m2o_value(..)
- _process_x2m_values(..)
"""
# We are using command with file template for that
file_template = self.env["cx.tower.file.template"].create(
{"name": "Test m2o", "reference": "test_m2o"}
)
file_template_values = file_template.with_context(
no_yaml_service_fields=True
)._prepare_record_for_yaml()
tag_doge_values = self.tag_doge.with_context(
no_yaml_service_fields=True
)._prepare_record_for_yaml()
tag_pepe_values = self.tag_pepe.with_context(
no_yaml_service_fields=True
)._prepare_record_for_yaml()
command = (
self.env["cx.tower.command"]
.create(
{
"name": "Command test m2o",
"action": "file_using_template",
"file_template_id": file_template.id,
"tag_ids": [(4, self.tag_doge.id), (4, self.tag_pepe.id)],
}
)
.with_context(explode_related_record=True)
) # and this is the actual trigger
# -- 1 --
# Record -> Yaml
# -- 1.1 --
# Many2one
result = command._process_relation_field_value(
field="file_template_id",
value=(command.file_template_id.id, command.file_template_id.name),
record_mode=True,
)
self.assertEqual(
result, file_template_values, "Reference was not resolved correctly"
)
# -- 1.2 --
# Many2many
result = command._process_relation_field_value(
field="tag_ids",
value=[self.tag_doge.id, self.tag_pepe.id],
record_mode=True,
)
self.assertEqual(len(result), 2, "Must be 2 records")
self.assertIn(tag_doge_values, result, "Record ID was not resolved correctly")
self.assertIn(tag_pepe_values, result, "Record ID was not resolved correctly")
# -- 2 --
# Yaml -> Record
# -- 2.1 --
# Many2one
result = command._process_relation_field_value(
field="file_template_id", value=file_template_values, record_mode=False
)
self.assertEqual(
result, file_template.id, "Record ID was not resolved correctly"
)
# -- 2.2 --
# Many2many
result = command._process_relation_field_value(
field="tag_ids", value=[tag_doge_values, tag_pepe_values], record_mode=False
)
self.assertEqual(len(result), 2, "Must be 2 records")
self.assertIn(
(4, self.tag_doge.id), result, "Record ID was not resolved correctly"
)
self.assertIn(
(4, self.tag_pepe.id), result, "Record ID was not resolved correctly"
)
# -- 3 --
# Yaml with non existing reference -> Record
file_template_values.update(
{
"name": "Very new name",
"reference": "such_much_not_reference",
"source": "server",
"file_type": "binary",
}
)
result = command._process_relation_field_value(
field="file_template_id", value=file_template_values, record_mode=False
)
# New record must be created
record = self.env["cx.tower.file.template"].browse(result)
self.assertEqual(
record.name, file_template_values["name"], "New record value doesn't match"
)
self.assertEqual(
record.reference,
file_template_values["reference"],
"New record value doesn't match",
)
self.assertEqual(
record.source,
file_template_values["source"],
"New record value doesn't match",
)
self.assertEqual(
record.file_type,
file_template_values["file_type"],
"New record value doesn't match",
)
# -- 4 --
# Yaml with no reference at all -> Record
values_with_no_references = {
"name": "Sorry no reference here",
"source": "tower",
"file_type": "binary",
}
result = command._process_relation_field_value(
field="file_template_id", value=values_with_no_references, record_mode=False
)
# New record must be created
record = self.env["cx.tower.file.template"].browse(result)
self.assertEqual(
record.name,
values_with_no_references["name"],
"New record value doesn't match",
)
self.assertEqual(
record.source,
values_with_no_references["source"],
"New record value doesn't match",
)
self.assertEqual(
record.file_type,
values_with_no_references["file_type"],
"New record value doesn't match",
)
# -- 5 --
# No record -> Yaml
result = command._process_relation_field_value(
field="file_template_id",
value=self.env["cx.tower.file.template"],
record_mode=True,
)
self.assertFalse(result, "Result must be 'False'")
def test_update_or_create_related_record(self):
"""Test if related record is updated or created correctly"""
# -- 1 --
# Update existing values
# We are using file template for that
FileTemplateModel = self.env["cx.tower.file.template"]
file_template = self.env["cx.tower.file.template"].create(
{"name": "Test m2o", "reference": "test_m2o"}
)
values_to_update = {"name": "Much new name"}
record = FileTemplateModel._update_or_create_related_record(
model=FileTemplateModel,
reference=file_template.reference,
values=values_to_update,
)
self.assertEqual(
record.name, values_to_update["name"], "Value was not updated properly"
)
self.assertEqual(record.id, file_template.id, "Same record must be updated")
# -- 2 --
# Reference not found. Must create a new record
values_to_update = {"name": "Doge file"}
record = FileTemplateModel._update_or_create_related_record(
model=FileTemplateModel,
reference="doge_file",
values=values_to_update,
create_immediately=True,
)
self.assertEqual(
record.name, values_to_update["name"], "Value was not updated properly"
)
self.assertNotEqual(record.id, file_template.id, "New record must be created")
# -- 2 --
# Reference not provided. Must create a new record
values_to_update = {"name": "Doge file"}
record = FileTemplateModel._update_or_create_related_record(
model=FileTemplateModel,
reference=False,
values=values_to_update,
create_immediately=True,
)
self.assertEqual(
record.name, values_to_update["name"], "Value was not updated properly"
)
self.assertNotEqual(record.id, file_template.id, "New record must be created")
@tagged("post_install", "-at_install")
def test_prepare_record_truncates_code_for_server_files(self):
"""Mixin must set code=False for cx.tower.file when source=='server'."""
File = self.env["cx.tower.file"]
srv_file = File.create(
{
"name": "srv.log",
"reference": "srvlog",
"source": "server",
"file_type": "text",
"server_dir": "/tmp",
"code": "BIG DATA",
}
)
rec = srv_file._prepare_record_for_yaml()
self.assertIn("code", rec)
self.assertFalse(rec["code"], "Expected code=False for server-sourced files")
@tagged("post_install", "-at_install")
def test_prepare_record_keeps_code_for_tower_files(self):
"""Mixin must keep code for cx.tower.file when source=='tower'."""
File = self.env["cx.tower.file"]
tw_file = File.create(
{
"name": "local.txt",
"reference": "localtxt",
"source": "tower",
"file_type": "text",
"server_dir": "/etc",
"code": "SMALL DATA",
}
)
rec = tw_file._prepare_record_for_yaml()
self.assertEqual(
rec["code"],
"SMALL DATA",
"Expected original code for tower-sourced files",
)

375
addons/cetmix_tower_yaml/tests/test_yaml_export_wizard.py
View File

@@ -1,375 +0,0 @@
# Copyright (C) 2024 Cetmix OÜ
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import base64
import yaml
from odoo.exceptions import AccessError, ValidationError
from odoo.addons.base.tests.common import BaseCommon
class TestYamlExportWizard(BaseCommon):
@classmethod
def setUpClass(cls, *args, **kwargs):
super().setUpClass(*args, **kwargs)
# Used to ensure that the file header
# is present in the YAML code
cls.file_header = """
# This file is generated with Cetmix Tower.
# Details and documentation: https://cetmix.com/tower
"""
# Create a command
cls.TowerCommand = cls.env["cx.tower.command"]
cls.command_test_wizard = cls.TowerCommand.create(
{
"reference": "test_command_from_yaml",
"name": "Test Command From Yaml",
"code": "echo 'Test Command From Yaml'",
}
)
cls.command_test_wizard_2 = cls.TowerCommand.create(
{
"reference": "test_command_from_yaml_2",
"name": "Test Command From Yaml 2",
"code": "echo 'Test Command From Yaml 2'",
}
)
# Create a flight plan
cls.FlightPlan = cls.env["cx.tower.plan"]
cls.flight_plan_test_wizard = cls.FlightPlan.create(
{
"name": "Test Flight Plan From Yaml",
"line_ids": [
(
0,
0,
{
"command_id": cls.command_test_wizard.id,
},
)
],
}
)
# Create a server template
cls.ServerTemplate = cls.env["cx.tower.server.template"]
cls.server_template_test_wizard = cls.ServerTemplate.create(
{
"name": "Test Server Template From Yaml",
"flight_plan_id": cls.flight_plan_test_wizard.id,
}
)
# Create a wizard and trigger onchange
cls.YamlExportWizard = cls.env["cx.tower.yaml.export.wiz"]
cls.test_wizard = cls.YamlExportWizard.with_context(
active_model="cx.tower.server.template",
active_ids=[cls.server_template_test_wizard.id],
).create({})
cls.test_wizard.onchange_explode_child_records()
def test_user_without_export_group_cannot_export(self):
"""Test if user without export group cannot export"""
# Tower manager user without export group
self.user_yaml_export = self.env["res.users"].create(
{
"name": "No Yaml Export User",
"login": "no_yaml_export_user",
"groups_id": [
(4, self.env.ref("cetmix_tower_server.group_manager").id)
],
}
)
with self.assertRaises(AccessError):
self.test_wizard.with_user(self.user_yaml_export).read([])
def test_yaml_export_wizard_yaml_generation(self):
"""Test code generation of YAML export wizard."""
wizard_yaml = """
# This file is generated with Cetmix Tower.
# Details and documentation: https://cetmix.com/tower
cetmix_tower_yaml_version: 1
records:
- cetmix_tower_model: command
access_level: manager
reference: test_command_from_yaml
name: Test Command From Yaml
action: ssh_command
allow_parallel_run: false
note: false
path: false
code: echo 'Test Command From Yaml'
server_status: false
no_split_for_sudo: false
if_file_exists: skip
disconnect_file: false
- cetmix_tower_model: command
access_level: manager
reference: test_command_from_yaml_2
name: Test Command From Yaml 2
action: ssh_command
allow_parallel_run: false
note: false
path: false
code: echo 'Test Command From Yaml 2'
server_status: false
no_split_for_sudo: false
if_file_exists: skip
disconnect_file: false
"""
# -- 1 --
# Test with two commands
context = {
"default_explode_child_records": True,
"default_remove_empty_values": True,
"active_model": "cx.tower.command",
"active_ids": [self.command_test_wizard.id, self.command_test_wizard_2.id],
}
wizard = self.YamlExportWizard.with_context(context).create({}) # pylint: disable=context-overridden # new need a new clean context
wizard.onchange_explode_child_records()
self.assertEqual(wizard.yaml_code, wizard_yaml)
def test_yaml_export_wizard(self):
"""Test the YAML export wizard."""
# -- 1 --
# Test wizard action
result = self.test_wizard.action_generate_yaml_file()
self.assertEqual(
result["type"], "ir.actions.act_window", "Action should be a window"
)
self.assertEqual(
result["res_model"],
"cx.tower.yaml.export.wiz.download",
"Result model should be the download wizard",
)
self.assertTrue(result["res_id"], "Wizard should have been created")
# -- 2 --
# Ensure download wizard file name is generated
# based on the record reference
download_wizard = self.env["cx.tower.yaml.export.wiz.download"].browse(
result["res_id"]
)
self.assertEqual(
download_wizard.yaml_file_name,
f"server_template_{self.server_template_test_wizard.reference}.yaml",
"YAML file name should be generated based on record reference",
)
# -- 3 --
# Decode YAML file and check if it's valid
yaml_file_content = base64.decodebytes(download_wizard.yaml_file).decode(
"utf-8"
)
self.assertEqual(
yaml_file_content,
self.test_wizard.yaml_code,
"YAML file content should be the same as the original YAML code",
)
# -- 4 --
# Test if empty YAML code is handled correctly
self.test_wizard.yaml_code = ""
with self.assertRaises(ValidationError):
self.test_wizard.action_generate_yaml_file()
def test_reference_object_uniqueness(self):
"""
Ensure each reference is exported as a full object only once
(other times only as ref).
"""
# Prepare YAML export for flight_plan with two same commands
self.flight_plan_test_wizard.line_ids = [
(0, 0, {"command_id": self.command_test_wizard.id}),
(0, 0, {"command_id": self.command_test_wizard.id}),
]
# Prepare YAML code
self.test_wizard.onchange_explode_child_records()
yaml_data = yaml.safe_load(self.test_wizard.yaml_code)
# reference counters
ref_full = set()
ref_refs = set()
# Recursively walk through the YAML data and count references
def walk(obj):
if isinstance(obj, dict):
ref = obj.get("reference")
# dict only with "reference" = ref, otherwise — full object
if ref:
if list(obj.keys()) == ["reference"]:
ref_refs.add(ref)
else:
ref_full.add(ref)
for v in obj.values():
walk(v)
elif isinstance(obj, list):
for v in obj:
walk(v)
# Walk through the YAML data
walk(yaml_data["records"])
# Each reference as a full object — only once
for ref in ref_full:
self.assertEqual(
list(ref_full).count(ref),
1,
f"Reference '{ref}' appears as a full object more than once",
)
# Check that no full objects appear more than once
self.assertEqual(
len(ref_full),
len(set(ref_full)),
"Some full objects appear more than once",
)
# Check that for each ref there is no only reference, but no full object
for ref in ref_refs:
self.assertIn(
ref,
ref_full,
f"Reference '{ref}' is used only as a reference, "
"but no full object present",
)
def test_export_required_model_name_in_yaml(self):
"""
Test that the model name is required in the YAML file for each record
"""
# create a command to run flight plan
command_run_flight_plan = self.TowerCommand.create(
{
"name": "Run Flight Plan",
"action": "plan",
"flight_plan_id": self.flight_plan_test_wizard.id,
}
)
# export 2 commands: command_run_flight_plan and command_test_wizard
wizard = self.YamlExportWizard.with_context(
active_model="cx.tower.command",
active_ids=[command_run_flight_plan.id, self.command_test_wizard.id],
).create({})
wizard.onchange_explode_child_records()
yaml_data = yaml.safe_load(wizard.yaml_code)
# check that the model name is present in the YAML file for each record
for record in yaml_data["records"]:
self.assertIn("cetmix_tower_model", record)
def test_default_yaml_file_name_is_used(self):
"""
Wizard should pre-fill `yaml_file_name` with the auto-generated
value that ends with '.yaml' and contains the model prefix.
"""
wiz = self.YamlExportWizard.with_context(
active_model="cx.tower.command",
active_ids=[self.command_test_wizard.id],
).create({})
default_name = wiz.yaml_file_name
self.assertFalse(
default_name.endswith(".yaml"),
"Default file name must NO have .yaml suffix",
)
self.assertIn(
"command_",
default_name,
"Default file name should include model prefix",
)
def test_yaml_file_name_is_auto_fixed(self):
"""
When the user assigns an invalid name, wizard should auto-sanitise
it to a safe *basename* (lowercase, underscores, no extension).
"""
wiz = self.YamlExportWizard.with_context(
active_model="cx.tower.command",
active_ids=[self.command_test_wizard.id],
).create({})
# user enters a 'dirty' name with spaces, capitals, symbols
wiz.write({"yaml_file_name": "My File!@# .YAML"})
# write() override strips to a basename WITHOUT '.yaml'
self.assertEqual(
wiz.yaml_file_name,
"my_file",
"Wizard field must hold only the cleaned basename, without extension",
)
def test_action_generate_appends_extension(self):
"""
When generating the download record, the system must append
the `.yaml` extension to the sanitized basename.
"""
wiz = self.YamlExportWizard.with_context(
active_model="cx.tower.command",
active_ids=[self.command_test_wizard.id],
).create({})
wiz.onchange_explode_child_records()
act = wiz.action_generate_yaml_file()
download = self.env["cx.tower.yaml.export.wiz.download"].browse(act["res_id"])
self.assertTrue(download.yaml_file_name.endswith(".yaml"))
def test_custom_requires_text(self):
"""Creating a template with license 'custom' but no text must fail"""
with self.assertRaises(ValidationError):
self.env["cx.tower.yaml.manifest.tmpl"].create(
{
"name": "Bad Manifest",
"license": "custom",
}
)
tmpl_ok = self.env["cx.tower.yaml.manifest.tmpl"].create(
{
"name": "Good Manifest",
"license": "custom",
"license_text": "Custom license terms",
}
)
self.assertEqual(tmpl_ok.license, "custom")
self.assertEqual(tmpl_ok.license_text, "Custom license terms")
with self.assertRaises(ValidationError):
self.env["cx.tower.yaml.manifest.tmpl"].create(
{
"name": "Bad Manifest 2",
"license": "custom",
"license_text": " ",
}
)
def test_wizard_resets_price_on_license_change(self):
"""Wizard must reset price/currency when license changes away from 'custom'"""
wiz = self.YamlExportWizard.new(
{
"manifest_license": "custom",
"manifest_price": 42.0,
"manifest_currency": "EUR",
}
)
wiz.manifest_license = "agpl-3"
wiz._onchange_manifest_license()
self.assertEqual(wiz.manifest_price, 0.0)
self.assertFalse(wiz.manifest_currency)
wiz.manifest_price = 7.5
wiz.manifest_currency = "USD"
wiz.manifest_license = "custom"
wiz._onchange_manifest_license()
self.assertEqual(wiz.manifest_price, 7.5)
self.assertEqual(wiz.manifest_currency, "USD")

703
addons/cetmix_tower_yaml/tests/test_yaml_import_wizard.py
View File

@@ -1,703 +0,0 @@
# Copyright (C) 2024 Cetmix OÜ
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
import base64
import yaml
from odoo import _
from odoo.exceptions import ValidationError
from odoo.tests import TransactionCase
from odoo.tools import mute_logger
class TestTowerYamlImportWizUpload(TransactionCase):
"""Test Tower YAML Import Wizard Upload"""
@classmethod
def setUpClass(cls):
super().setUpClass()
# Variables
cls.Variable = cls.env["cx.tower.variable"]
cls.variable_yaml_test = cls.Variable.create(
{"name": "YAML Test", "reference": "yaml_test"}
)
cls.variable_yaml_url = cls.Variable.create(
{"name": "YAML URL", "reference": "yaml_url"}
)
# Tags
cls.Tag = cls.env["cx.tower.tag"]
cls.tag_yaml_test = cls.Tag.create(
{"name": "YAML Test", "reference": "yaml_test"}
)
cls.tag_another_yaml_test = cls.Tag.create(
{"name": "Another YAML Test", "reference": "another_yaml_test"}
)
# Commands
cls.Command = cls.env["cx.tower.command"]
cls.command_yaml_test = cls.Command.create(
{"name": "Test Yaml Command", "reference": "test_yaml_command"}
)
# Flight Plan
cls.FlightPlan = cls.env["cx.tower.plan"]
cls.flight_plan_yaml_test = cls.FlightPlan.create(
{
"name": "Test Yaml Flight Plan",
"reference": "test_yaml_flight_plan",
"line_ids": [
(
0,
0,
{
"condition": False,
"use_sudo": False,
"command_id": cls.command_yaml_test.id,
},
),
],
}
)
# Create Server Template used for testing
cls.server_template_yaml_test = cls.env["cx.tower.server.template"].create(
{
"name": "Test Server Template",
"tag_ids": [
(4, cls.tag_yaml_test.id),
(4, cls.tag_another_yaml_test.id),
],
"variable_value_ids": [
(
0,
0,
{
"variable_id": cls.variable_yaml_test.id,
"value_char": "Some Test Value",
},
),
(
0,
0,
{
"variable_id": cls.variable_yaml_url.id,
"value_char": "https://cetmix.com",
},
),
],
"flight_plan_id": cls.flight_plan_yaml_test.id,
}
)
# Server Logs
cls.ServerLog = cls.env["cx.tower.server.log"]
cls.server_log_yaml_test = cls.ServerLog.create(
{
"name": "Test Server Log",
"reference": "test_server_log",
"command_id": cls.command_yaml_test.id,
"log_type": "command",
"server_template_id": cls.server_template_yaml_test.id,
}
)
# Create an export wizard and generate YAML code
context = {
"active_model": "cx.tower.server.template",
"active_ids": [cls.server_template_yaml_test.id],
}
cls.export_wizard = (
cls.env["cx.tower.yaml.export.wiz"].with_context(context).create({}) # pylint: disable=context-overridden # new need a new clean context
)
cls.export_wizard.onchange_explode_child_records()
cls.export_wizard.action_generate_yaml_file()
cls.yaml_code = cls.export_wizard.yaml_code
cls.yaml_file = base64.b64encode(cls.yaml_code.encode("utf-8"))
# YAML import upload wizard
cls.YamlImportWizUpload = cls.env["cx.tower.yaml.import.wiz.upload"]
cls.yaml_upload_wizard = cls.YamlImportWizUpload.create(
{"yaml_file": cls.yaml_file, "file_name": "test_yaml_file.yaml"}
)
# YAML import wizard
cls.import_wizard_action = cls.yaml_upload_wizard.action_import_yaml()
cls.import_wizard = cls.env[cls.import_wizard_action["res_model"]].browse(
cls.import_wizard_action["res_id"]
)
cls.import_wizard.if_record_exists = "update"
def test_extract_yaml_data(self):
"""Test extract YAML data from file"""
# -- 1 --
# Test if YAML file is valid
extracted_yaml_data = self.yaml_upload_wizard._extract_yaml_data()
self.assertEqual(
extracted_yaml_data,
self.yaml_code,
"YAML code is not extracted correctly",
)
# -- 2 --
# Test if invalid model is handled properly
# Replace model name with invalid model
self.invalid_yaml_code = self.yaml_code.replace(
"server_template", "invalid_model"
)
self.invalid_yaml_file = base64.b64encode(
self.invalid_yaml_code.encode("utf-8")
)
self.yaml_upload_wizard.yaml_file = self.invalid_yaml_file
with self.assertRaises(ValidationError) as e:
self.yaml_upload_wizard._extract_yaml_data()
self.assertEqual(
str(e.exception),
_("'invalid_model' is not a valid model"),
"Exception message does not match",
)
# -- 3 --
# Test if non YAML supported model is handled properly
# Replace model name with non YAML supported model
self.non_yaml_supported_yaml_code = self.yaml_code.replace(
"server_template", "command_run_wizard"
)
self.non_yaml_supported_yaml_file = base64.b64encode(
self.non_yaml_supported_yaml_code.encode("utf-8")
)
self.yaml_upload_wizard.yaml_file = self.non_yaml_supported_yaml_file
with self.assertRaises(ValidationError) as e:
self.yaml_upload_wizard._extract_yaml_data()
self.assertEqual(
str(e.exception),
_("Model 'command_run_wizard' does not support YAML import"),
"Exception message does not match",
)
# -- 4 --
# Test if YAML that is not a dictionary is handled properly
self.invalid_yaml_file = base64.b64encode(b"Invalid YAML file")
self.yaml_upload_wizard.yaml_file = self.invalid_yaml_file
with self.assertRaises(ValidationError) as e:
self.yaml_upload_wizard._extract_yaml_data()
self.assertEqual(
str(e.exception),
_("Yaml file doesn't contain valid data"),
"Exception message does not match",
)
# -- 5 --
# Test if TypeError is handled properly
self.non_unicode_yaml_file = base64.b64encode(b"\x80")
self.yaml_upload_wizard.yaml_file = self.non_unicode_yaml_file
with self.assertRaises(ValidationError) as e:
self.yaml_upload_wizard._extract_yaml_data()
self.assertEqual(
str(e.exception),
_("YAML file cannot be decoded properly"),
"Exception message does not match",
)
# -- 6 --
# Test if YAML file is empty
self.empty_yaml_file = ""
self.yaml_upload_wizard.yaml_file = self.empty_yaml_file
with self.assertRaises(ValidationError) as e:
self.yaml_upload_wizard._extract_yaml_data()
self.assertEqual(
str(e.exception),
_("File is empty"),
"Exception message does not match",
)
# -- 7 --
# Test if YAML file with unsupported YAML version is handled properly
yaml_with_unsupported_version = self.yaml_code.replace(
f"cetmix_tower_yaml_version: {self.FlightPlan.CETMIX_TOWER_YAML_VERSION}",
f"cetmix_tower_yaml_version: {self.FlightPlan.CETMIX_TOWER_YAML_VERSION + 1}", # noqa: E501
)
self.unsupported_yaml_version_yaml_file = base64.b64encode(
yaml_with_unsupported_version.encode("utf-8")
)
self.yaml_upload_wizard.yaml_file = self.unsupported_yaml_version_yaml_file
with self.assertRaises(ValidationError) as e:
self.yaml_upload_wizard._extract_yaml_data()
self.assertEqual(
str(e.exception),
_(
"YAML version is higher than version"
" supported by your Cetmix Tower instance."
" %(code_version)s > %(tower_version)s",
code_version=self.FlightPlan.CETMIX_TOWER_YAML_VERSION + 1,
tower_version=self.FlightPlan.CETMIX_TOWER_YAML_VERSION,
),
"Exception message does not match",
)
# -- 8 --
# Test YAML file with no records
self.import_wizard.yaml_code = "cetmix_tower_yaml_version: 1"
with self.assertRaises(ValidationError) as e:
self.import_wizard.action_import_yaml()
self.assertEqual(
str(e.exception),
_("YAML file doesn't contain any records"),
"Exception message does not match",
)
def test_action_import_yaml_skip_if_exists(self):
"""Test YAML import wizard action when skipping an existing record"""
self.import_wizard.if_record_exists = "skip"
# Run import wizard action
import_wizard_result_action = self.import_wizard.action_import_yaml()
# Test if action is composed properly
self.assertEqual(
import_wizard_result_action["type"],
"ir.actions.client",
"Import wizard action type is not correct",
)
self.assertEqual(
import_wizard_result_action["tag"],
"display_notification",
"Import wizard action tag is not correct",
)
self.assertEqual(
import_wizard_result_action["params"]["title"],
_("Record Import"),
"Import wizard action title is not correct",
)
self.assertEqual(
import_wizard_result_action["params"]["message"],
_("No records were created or updated"),
"Import wizard action message is not correct",
)
self.assertEqual(
import_wizard_result_action["params"]["sticky"],
True,
"Import wizard action sticky is not correct",
)
self.assertEqual(
import_wizard_result_action["params"]["type"],
"warning",
"Import wizard action type is not correct",
)
def test_action_import_yaml_update_existing_record(self):
"""Test YAML import wizard action when updating an existing record"""
# -- 1 --
# Test if new import wizard record is created properly
self.assertEqual(
self.import_wizard_action["res_model"],
"cx.tower.yaml.import.wiz",
"Import wizard action model is not correct",
)
self.assertEqual(
self.import_wizard_action["view_mode"],
"form",
"Import wizard action view mode is not correct",
)
# -- 2 --
# Modify Server Template name and variable value
self.import_wizard.yaml_code = self.import_wizard.yaml_code.replace(
"name: Test Server Template",
"name: Updated Test Server Template",
).replace(
"value_char: Some Test Value",
"value_char: Updated Test Value",
)
variable_value_to_update = (
self.server_template_yaml_test.variable_value_ids.filtered(
lambda v: v.value_char == "Some Test Value"
)
)
# Run import wizard action another time
import_wizard_result_action = self.import_wizard.action_import_yaml()
# -- 3 --
# Test if record is updated properly
self.assertEqual(
import_wizard_result_action["res_model"],
"cx.tower.server.template",
"Import wizard action model is not correct",
)
self.assertEqual(
import_wizard_result_action["domain"],
[("id", "in", self.server_template_yaml_test.ids)],
"ID must match existing record ID",
)
self.assertEqual(
self.server_template_yaml_test.name,
"Updated Test Server Template",
"Record is not updated properly",
)
self.assertEqual(
variable_value_to_update.value_char,
"Updated Test Value",
"Variable value is not updated properly",
)
# -- 4 --
# Test if server log remains the same
self.assertEqual(
len(self.server_template_yaml_test.server_log_ids),
1,
"Server Log must remain the same",
)
self.assertEqual(
self.server_log_yaml_test.id,
self.server_template_yaml_test.server_log_ids.id,
"Server Log must remain the same",
)
def test_action_import_yaml_create_new_record(self):
"""Test YAML import wizard action when creating a new record"""
self.import_wizard.if_record_exists = "create"
with mute_logger("odoo.addons.cetmix_tower_yaml.models.cx_tower_yaml_mixin"):
import_wizard_result_action = self.import_wizard.action_import_yaml()
# -- 1 --
# Test if new record is created instead of updating existing one
self.assertEqual(
import_wizard_result_action["res_model"],
"cx.tower.server.template",
"Import wizard action model is not correct",
)
self.assertNotEqual(
import_wizard_result_action["domain"],
f"[('id', '=', {self.server_template_yaml_test.ids})]",
"ID must not match existing record ID",
)
# -- 2 --
# Ensure that existing flight plan is used instead of creating a new one
new_server_template = self.env[import_wizard_result_action["res_model"]].search(
import_wizard_result_action["domain"]
)
self.assertEqual(
new_server_template.flight_plan_id,
self.flight_plan_yaml_test,
"Existing flight plan must be used",
)
# -- 3 --
# Ensure that existing tags are used instead of creating new ones
for tag in self.server_template_yaml_test.tag_ids:
self.assertIn(
tag,
new_server_template.tag_ids,
"Existing tag must be used",
)
# -- 4 --
# Ensure that new variable values are created
for variable_value in self.server_template_yaml_test.variable_value_ids:
self.assertNotIn(
variable_value,
new_server_template.variable_value_ids,
"New variable value must be created instead of updating existing one",
)
# -- 5 --
# Test if server log is created instead of updated
for server_log in self.server_template_yaml_test.server_log_ids:
self.assertNotIn(
server_log,
new_server_template.server_log_ids,
"New Server Log must be created instead of updating existing one",
)
def test_extract_secret_names(self):
"""Test extract secret names from YAML data"""
# NB: this is not a real model, it's just for testing
yaml_code = """cetmix_tower_yaml_version: 1
records:
- cetmix_tower_model: test_model
access_level: manager
reference: such_much_test_record
name: Such Much Command
action: file_using_template
allow_parallel_run: false
note: Just a note
os_ids: false
tag_ids: false
path: false
file_template_id: false
flight_plan_id: false
code: false
variable_ids: false
secret_ids: false
ssh_key_id:
reference: test_ssh_key
name: Test SSH Key
key_type: k
note: false
- cetmix_tower_model: another_test_model
reference: such_much_test_record_2
name: Such Much Test Record 2
note: Just a note 2
ssh_key_id:
reference: test_ssh_key
name: Test SSH Key
key_type: k
note: false
secret_ids:
- reference: secret_2
name: Secret 2
key_type: s
note: false
- reference: secret_3
name: Secret 3
key_type: s
note: false
- cetmix_tower_model: another_test_model
reference: such_much_test_record_3
name: Such Much Test Record 3
note: Just a note 3
ssh_key_id:
reference: another_ssh_key
name: Another SSH Key
sub_record:
reference: such_much_test_record_4
name: Such Much Test Record 4
note: Just a note 4
secret_ids:
- reference: secret_1
name: Secret 3
key_type: s
note: false
- reference: secret_2
name: Secret 4
key_type: s
note: false
file_template_id:
reference: my_custom_test_template
name: Such much demo
source: tower
file_type: text
server_dir: /var/log/my/files
file_name: much_logs.txt
keep_when_deleted: false
tag_ids: false
note: Hey!
code: false
variable_ids: false
secret_ids: false
flight_plan_id: false
code: false
variable_ids: false
secret_ids:
- reference: secret_1
name: Secret 1
key_type: s
note: false
- reference: secret_2
name: Secret 2
key_type: s
note: false
"""
secret_list = self.env["cx.tower.yaml.import.wiz"]._extract_secret_names(
yaml.safe_load(yaml_code)
)
# We expect 6 secrets in the list:
# 2 keys: 'Test SSH Key', 'Another SSH Key'
# 4 secrets: 'Secret 3', 'Secret 4', 'Secret 1', 'Secret 2'
self.assertEqual(len(secret_list), 6, "Secret list length is not correct")
self.assertIn("Test SSH Key", secret_list, "Key is not in the list")
self.assertIn("Another SSH Key", secret_list, "Key is not in the list")
self.assertIn("Secret 3", secret_list, "Key is not in the list")
self.assertIn("Secret 4", secret_list, "Key is not in the list")
self.assertIn("Secret 1", secret_list, "Key is not in the list")
self.assertIn("Secret 2", secret_list, "Key is not in the list")
def test_extract_secret_names_with_key_id(self):
"""Test extract secret names when secrets are nested under key_id"""
yaml_code = """cetmix_tower_yaml_version: 1
records:
- cetmix_tower_model: test_model
reference: rec_1
name: Test Record
secret_ids:
- key_id:
reference: secret_1
name: Nested Secret 1
- key_id:
reference: secret_2
name: Nested Secret 2
ssh_key_id:
name: SSH Key Nested
"""
secret_list = self.env["cx.tower.yaml.import.wiz"]._extract_secret_names(
yaml.safe_load(yaml_code)
)
# We expect 3 secrets total:
# - SSH Key Nested (from ssh_key_id)
# - Nested Secret 1
# - Nested Secret 2
self.assertCountEqual(
secret_list,
["Nested Secret 1", "Nested Secret 2", "SSH Key Nested"],
"Unexpected secrets extracted for nested structure",
)
def test_create_records_different_models(self):
"""Test create records with different models"""
yaml_code = """cetmix_tower_yaml_version: 1
records:
- cetmix_tower_model: command
access_level: manager
reference: much_much_command
name: Much Much Command
action: file_using_template
allow_parallel_run: false
note: Just a note
os_ids: false
tag_ids: false
path: false
file_template_id: false
flight_plan_id: false
code: false
variable_ids: false
secret_ids: false
ssh_key_id:
reference: test_ssh_key
name: Test SSH Key
key_type: k
note: false
- cetmix_tower_model: server_template
reference: wow_much_server_template
name: Wow Much Server Template
note: Just a note 2
- cetmix_tower_model: tag
reference: such_much_tag
name: Such Much Tag
"""
# Create a new command record
self.import_wizard.if_record_exists = "update"
self.import_wizard.yaml_code = yaml_code
action = self.import_wizard.action_import_yaml()
# Check if action is composed properly
self.assertEqual(
action["type"],
"ir.actions.client",
"Import wizard action type is not correct",
)
self.assertEqual(
action["tag"],
"display_notification",
"Import wizard action tag is not correct",
)
self.assertEqual(
action["params"]["title"],
_("Record Import"),
"Import wizard action title is not correct",
)
self.assertEqual(
action["params"]["type"],
"success",
"Import wizard action type is not correct",
)
self.assertEqual(
action["params"]["sticky"],
True,
"Import wizard action sticky is not correct",
)
# Check command
self.assertTrue(
self.Command.get_by_reference("much_much_command"),
"Command must be created",
)
# Check server template
self.assertTrue(
self.env["cx.tower.server.template"].get_by_reference(
"wow_much_server_template"
),
"Server template must be created",
)
# Check tag
self.assertTrue(
self.Tag.get_by_reference("such_much_tag"), "Tag must be created"
)
def test_yaml_import_server_without_password(self):
"""Wizard should import server without ssh_password."""
yaml_code = (
"cetmix_tower_yaml_version: 1\n"
"records:\n"
"- reference: srv_nopass\n"
" cetmix_tower_model: server\n"
" name: YAML NoPass\n"
" ssh_auth_mode: p\n"
" ssh_username: root\n"
" ip_v4_address: 10.0.0.3\n"
)
wiz = self.env["cx.tower.yaml.import.wiz"].create(
{
"yaml_code": yaml_code,
"if_record_exists": "create",
}
)
wiz.action_import_yaml()
srv = self.env["cx.tower.server"].get_by_reference("srv_nopass")
self.assertTrue(srv, "Server was not created")
self.assertFalse(
srv._get_secret_value("ssh_password"),
"ssh_password must stay empty after import",
)
def test_orm_create_server_requires_password(self):
"""Creating a server via ORM/UI must fail when ssh_password is missing."""
with self.assertRaises(ValidationError) as err:
self.env["cx.tower.server"].create(
{
"reference": "srv_ui",
"name": "UI NoPass",
"ssh_auth_mode": "p",
"ssh_username": "root",
"ip_v4_address": "10.0.0.2",
}
)
self.assertIn("Please provide SSH password", str(err.exception))
def test_yaml_import_server_with_skip_ssh_check(self):
"""Explicit skip_ssh_settings_check also bypasses password validation."""
yaml_code = (
"cetmix_tower_yaml_version: 1\n"
"records:\n"
"- reference: srv_skip\n"
" cetmix_tower_model: server\n"
" name: YAML Skip Check\n"
" ssh_auth_mode: p\n"
" ssh_username: root\n"
" ip_v4_address: 10.0.0.4\n"
)
wiz = self.env["cx.tower.yaml.import.wiz"].create(
{
"yaml_code": yaml_code,
"if_record_exists": "create",
}
)
wiz.with_context(skip_ssh_settings_check=True).action_import_yaml()
srv = self.env["cx.tower.server"].get_by_reference("srv_skip")
self.assertTrue(
srv, "Server must be created when skip_ssh_settings_check is set"
)