Compare commits
2 Commits
v2026.02.0
...
v2026.02.0
| Author | SHA1 | Date | |
|---|---|---|---|
| 9e28004d6c | |||
| 8b3584fa9e |
31
test_inventory_keys.py
Normal file
31
test_inventory_keys.py
Normal file
@@ -0,0 +1,31 @@
|
|||||||
|
import unittest
|
||||||
|
from wif2ansible.models import Server, Flow
|
||||||
|
from wif2ansible.inventory import generate_inventory
|
||||||
|
|
||||||
|
class TestInventoryKeys(unittest.TestCase):
|
||||||
|
def test_inventory_keys_are_hostnames(self):
|
||||||
|
# Create a server with Ref, Hostname, IP
|
||||||
|
s1 = Server(reference="SERVER_REF_01", hostname="server01", ip_address="192.168.1.10", platform="windows")
|
||||||
|
|
||||||
|
# Create a flow matching this server
|
||||||
|
f1 = Flow(flow_id="1", source_ip="192.168.1.10", destination_ip="10.0.0.1", ports=[80])
|
||||||
|
|
||||||
|
servers = {"SERVER_REF_01": s1}
|
||||||
|
flows = [f1]
|
||||||
|
|
||||||
|
inventory = generate_inventory(servers, flows)
|
||||||
|
|
||||||
|
# Verify stricture
|
||||||
|
hosts = inventory['all']['hosts']
|
||||||
|
|
||||||
|
# Key should be REFERENCE "SERVER_REF_01" (or hostname/ip fallback)
|
||||||
|
self.assertIn("SERVER_REF_01", hosts)
|
||||||
|
self.assertNotIn("192.168.1.10", hosts)
|
||||||
|
|
||||||
|
# Check variables
|
||||||
|
host_vars = hosts["SERVER_REF_01"]
|
||||||
|
self.assertEqual(host_vars['ansible_host'], "192.168.1.10")
|
||||||
|
self.assertEqual(host_vars['ansible_connection'], "winrm")
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
||||||
@@ -2,7 +2,7 @@ import openpyxl
|
|||||||
from openpyxl.worksheet.worksheet import Worksheet
|
from openpyxl.worksheet.worksheet import Worksheet
|
||||||
from typing import List, Dict, Tuple, Optional
|
from typing import List, Dict, Tuple, Optional
|
||||||
from .models import Server, Flow
|
from .models import Server, Flow
|
||||||
from .parsers import parse_ports, parse_ip, clean_header
|
from .parsers import parse_ports, parse_ip, clean_header, clean_reference
|
||||||
|
|
||||||
from openpyxl.utils import get_column_letter
|
from openpyxl.utils import get_column_letter
|
||||||
|
|
||||||
@@ -130,7 +130,7 @@ def read_servers(filename: str) -> Dict[str, Server]:
|
|||||||
|
|
||||||
s = Server(
|
s = Server(
|
||||||
reference=ref,
|
reference=ref,
|
||||||
hostname=ref,
|
hostname=clean_reference(ref),
|
||||||
platform=plat,
|
platform=plat,
|
||||||
ip_address=ip_addr,
|
ip_address=ip_addr,
|
||||||
production_ip=prod_ip_addr
|
production_ip=prod_ip_addr
|
||||||
|
|||||||
@@ -61,8 +61,8 @@ def generate_inventory(servers: Dict[str, Server], flows: List[Flow]) -> Dict[st
|
|||||||
match_count += 1
|
match_count += 1
|
||||||
|
|
||||||
# Prepare host entry if new
|
# Prepare host entry if new
|
||||||
# We use the IP as the key in inventory 'hosts'
|
# We use the Reference/Hostname as the key in inventory 'hosts'
|
||||||
host_key = server.ip_address
|
host_key = server.reference or server.hostname or server.ip_address
|
||||||
|
|
||||||
if host_key not in inventory_hosts:
|
if host_key not in inventory_hosts:
|
||||||
host_vars = server.get_ansible_vars()
|
host_vars = server.get_ansible_vars()
|
||||||
|
|||||||
@@ -23,6 +23,9 @@ class Server:
|
|||||||
# Default ssh is usually fine, but being explicit doesn't hurt
|
# Default ssh is usually fine, but being explicit doesn't hurt
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
if self.ip_address:
|
||||||
|
vars['ansible_host'] = self.ip_address
|
||||||
|
|
||||||
return vars
|
return vars
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
|||||||
@@ -57,6 +57,21 @@ def parse_ports(port_str: str) -> List[int]:
|
|||||||
|
|
||||||
return sorted(list(ports))
|
return sorted(list(ports))
|
||||||
|
|
||||||
|
def clean_reference(ref: str) -> str:
|
||||||
|
"""
|
||||||
|
Cleans a server reference string.
|
||||||
|
Specifically removes 'SRV###' type prefixes if present.
|
||||||
|
Example: 'SRV123 MyServer' -> 'MyServer'
|
||||||
|
"""
|
||||||
|
if not ref:
|
||||||
|
return ""
|
||||||
|
|
||||||
|
s = str(ref)
|
||||||
|
# Remove SRV followed by digits and whitespace
|
||||||
|
s = re.sub(r'SRV\d+\s*', '', s, flags=re.IGNORECASE)
|
||||||
|
# Remove leading/trailing whitespace
|
||||||
|
return s.strip()
|
||||||
|
|
||||||
def parse_ip(ip_str: str) -> List[str]:
|
def parse_ip(ip_str: str) -> List[str]:
|
||||||
"""Finds all IPv4 addresses in a string."""
|
"""Finds all IPv4 addresses in a string."""
|
||||||
if not ip_str:
|
if not ip_str:
|
||||||
|
|||||||
Reference in New Issue
Block a user