from __future__ import annotations import unittest from uvm.domain import PortRule from uvm.errors import ValidationError from uvm.validation import ( parse_cpu, parse_name, parse_password, parse_ports, parse_ram, parse_storage, parse_timeout, parse_username, ) class ValidationTests(unittest.TestCase): def test_storage_defaults_to_gib_and_supports_larger_and_fractional_units(self) -> None: for value, expected in ( ("30", 30720), ("30G", 30720), ("30GB", 30720), ("30GiB", 30720), (" 50 g ", 51200), ("30720M", 30720), ("30.5G", 31232), ("1T", 1048576), ("0.5TiB", 524288), ("30.0001G", 30721), ): with self.subTest(value=value): self.assertEqual(parse_storage(value), expected) def test_storage_rejects_subminimum_invalid_and_unrepresentable_capacity(self) -> None: for value in ("0", "29G", "29.999999G", "30719M", "30M", "", "-30G", "nan", "inf", "True", "30G garbage", "100000000000000000000000T"): with self.subTest(value=value), self.assertRaises(ValidationError): parse_storage(value) def test_parse_ram_uses_mib_by_default(self) -> None: self.assertEqual(parse_ram("512"), 512) self.assertEqual(parse_ram("1G"), 1024) self.assertEqual(parse_ram("512MiB"), 512) self.assertEqual(parse_ram("134217728B"), 128) def test_parse_ram_rejects_small_and_invalid_values(self) -> None: for value in ("127M", "nonsense", "-1G", "1T"): with self.subTest(value=value): with self.assertRaises(ValidationError): parse_ram(value) def test_parse_cpu_requires_a_positive_finite_value(self) -> None: self.assertEqual(parse_cpu("0.5"), 0.5) for value in ("0", "-1", "nan", "inf", "cpu"): with self.subTest(value=value): with self.assertRaises(ValidationError): parse_cpu(value) def test_guest_username_is_validated(self) -> None: self.assertEqual(parse_username("root"), "root") self.assertEqual(parse_username("app-user"), "app-user") for value in ("", "Root", "9user", "user name", "a" * 33): with self.subTest(value=value): with self.assertRaises(ValidationError): parse_username(value) def test_guest_password_rejects_empty_control_or_oversized_values(self) -> None: self.assertEqual( parse_password("correct horse battery staple"), "correct horse battery staple", ) for value in ("", "line\nbreak", "tab\tvalue", "a" * 129): with self.subTest(value=value): with self.assertRaises(ValidationError): parse_password(value) def test_names_accept_full_lowercase_names_up_to_63_characters(self) -> None: for name in ("a", "app-server", "worker2", "a" * 63, "vm-test", "vm-" + "g" * 32): with self.subTest(name=name): self.assertEqual(parse_name(name), name) def test_names_reject_invalid_format_case_and_length(self) -> None: for name in ( "", "App", "APP", "9app", "-app", "app_name", "app.name", "app/name", " app", "app ", "app\n", "a" * 64, "caf\u00e9", "app;id", ): with self.subTest(name=name), self.assertRaises(ValidationError): parse_name(name) def test_names_cannot_impersonate_a_generated_vm_id(self) -> None: for suffix in ("0" * 32, "0123456789abcdef" * 2, "f" * 32): with self.subTest(suffix=suffix): with self.assertRaisesRegex(ValidationError, "impersonate generated VM IDs"): parse_name("vm-" + suffix) for suffix in ("f" * 31, "f" * 33): self.assertEqual(parse_name("vm-" + suffix), "vm-" + suffix) def test_ports_accept_repeated_comma_lists_whitespace_and_case_insensitive_labels(self) -> None: values = ("443:https,22", " 80:HtTp ,\t8080:tcp ", "65535, 1") expected = ( PortRule(1), PortRule(22), PortRule(80, "HTTP"), PortRule(443, "HTTPS"), PortRule(8080), PortRule(65535), ) self.assertEqual(parse_ports(values), expected) self.assertEqual(parse_ports(list(values)), expected) def test_bare_ports_are_tcp_and_empty_sequences_do_not_add_defaults(self) -> None: self.assertEqual(parse_ports(("80,443",)), (PortRule(80, "TCP"), PortRule(443, "TCP"))) self.assertEqual(parse_ports([]), ()) self.assertEqual(parse_ports(()), ()) def test_identical_port_labels_are_deduplicated_across_options_and_comma_lists(self) -> None: self.assertEqual( parse_ports(("22", "22:tcp", "80:HTTP,80:hTtP", "443:HTTPS", "443:https")), (PortRule(22), PortRule(80, "HTTP"), PortRule(443, "HTTPS")), ) def test_conflicting_tcp_service_labels_are_rejected(self) -> None: for values in ( ("80", "80:HTTP"), ("80:HTTP,80:TCP",), ("80:HTTP", "80:HTTPS"), ("80:https,80:http",), ): with self.subTest(values=values): with self.assertRaisesRegex(ValidationError, "conflicting labels for TCP port 80"): parse_ports(values) def test_ports_require_strict_port_colon_label_syntax(self) -> None: for value in ( "", ",", "22,", ",22", "22,,80", "22, ,80", "TCP", "HTTP", "HTTPS", "TCP:80", "80/tcp", ":HTTP", "80:", "80::HTTP", "80 :HTTP", "80: HTTP", "80:HTTP S", "80:UDP", "80-81", "80.0", "+80", "0x50", "22 80", "80;id", ): with self.subTest(value=value), self.assertRaises(ValidationError): parse_ports((value,)) def test_parsed_ports_reject_out_of_range_numbers_and_non_integer_text(self) -> None: for value in ("-1", "0", "65536", "999999999999", "True", "False", "None", "1.5"): with self.subTest(value=value), self.assertRaises(ValidationError): parse_ports((value,)) def test_ports_reject_scalar_strings_instead_of_splitting_them_into_digits(self) -> None: for values in ("22", "", b"22"): with self.subTest(values=values), self.assertRaises(ValidationError): parse_ports(values) def test_ports_require_a_sequence_not_a_scalar_mapping_or_set(self) -> None: for values in (None, True, 22, {"22": "HTTP"}, {"22"}): with self.subTest(values=values), self.assertRaises(ValidationError): parse_ports(values) def test_ports_require_string_sequence_members(self) -> None: for value in (None, True, False, 22, 22.0, b"22", [], {}, PortRule(22)): with self.subTest(value=value), self.assertRaises(ValidationError): parse_ports(("80:HTTP", value)) def test_port_rules_require_real_integers_within_tcp_port_bounds(self) -> None: self.assertEqual(PortRule(1).protocol, "TCP") self.assertEqual(PortRule(65535, "https").protocol, "HTTPS") for port in (None, True, False, 0, -1, 65536, 22.0, "22", b"22", [], {}): with self.subTest(port=port), self.assertRaises(ValidationError): PortRule(port) def test_port_rule_labels_are_normalized_but_not_coerced_or_trimmed(self) -> None: self.assertEqual(PortRule(8080, "hTtP"), PortRule(8080, "HTTP")) for protocol in (None, True, 1, "", "UDP", "SCTP", " TCP", "TCP ", "tcp\n", [], {}): with self.subTest(protocol=protocol), self.assertRaises(ValidationError): PortRule(22, protocol) def test_timeout_accepts_zero_and_fractional_seconds(self) -> None: self.assertEqual(parse_timeout("0"), 0) self.assertEqual(parse_timeout("0.5"), 0.5) def test_timeout_rejects_negative_non_finite_or_invalid_values(self) -> None: for value in ("-1", "nan", "inf", "-inf", "", "seconds", None): with self.subTest(value=value), self.assertRaises(ValidationError): parse_timeout(value)