Configuration Management with Ansible

1import ansible_runner
2import yaml
3from pathlib import Path
4from typing import Dict, List, Optional, Any
5import logging
6from datetime import datetime
7import json
8
9logging.basicConfig(level=logging.INFO)
10logger = logging.getLogger(__name__)
11
12
13class AnsiblePlaybookRunner:
14    """
15    Execute Ansible playbooks programmatically using ansible-runner
16    """
17
18    def __init__(
19        self,
20        private_data_dir: Path = Path('./ansible'),
21        inventory: Optional[str] = None
22    ):
23        """
24        Initialize Ansible runner
25
26        Args:
27            private_data_dir: Directory containing playbooks and roles
28            inventory: Path to inventory file or inline inventory
29        """
30        self.private_data_dir = Path(private_data_dir)
31        self.inventory = inventory
32        self.results = []
33
34        # Ensure directories exist
35        self.private_data_dir.mkdir(parents=True, exist_ok=True)
36        (self.private_data_dir / 'project').mkdir(exist_ok=True)
37        (self.private_data_dir / 'inventory').mkdir(exist_ok=True)
38
39        logger.info(f"Ansible runner initialized: {self.private_data_dir}")
40
41    def run_playbook(
42        self,
43        playbook: str,
44        inventory: Optional[Dict] = None,
45        extravars: Optional<br/> = None,
46        limit: Optional[str] = None,
47        verbosity: int = 0
48    ) -> ansible_runner.Runner:
49        """
50        Execute Ansible playbook
51
52        Args:
53            playbook: Playbook filename or content
54            inventory: Inventory dictionary
55            extravars: Extra variables to pass
56            limit: Limit execution to specific hosts
57            verbosity: Verbosity level (0-4)
58
59        Returns:
60            Runner instance with execution results
61        """
62        logger.info(f"Executing playbook: {playbook}")
63
64        # Prepare inventory
65        if inventory:
66            inventory_data = inventory
67        elif self.inventory:
68            inventory_data = self.inventory
69        else:
70            inventory_data = {'all': {'hosts': ['localhost']}}
71
72        # Execute playbook
73        runner = ansible_runner.run(
74            private_data_dir=str(self.private_data_dir),
75            playbook=playbook,
76            inventory=inventory_data,
77            extravars=extravars or {},
78            limit=limit,
79            verbosity=verbosity,
80            quiet=False
81        )
82
83        # Store results
84        self.results.append({
85            'playbook': playbook,
86            'status': runner.status,
87            'rc': runner.rc,
88            'stats': runner.stats,
89            'timestamp': datetime.now().isoformat()
90        })
91
92        # Log results
93        if runner.status == 'successful':
94            logger.info(f"✓ Playbook completed successfully")
95            logger.info(f"  Stats: {json.dumps(runner.stats, indent=2)}")
96        else:
97            logger.error(f"✗ Playbook failed: {runner.status}")
98
99        return runner
100
101    def run_adhoc_command(
102        self,
103        module: str,
104        module_args: Optional[Dict] = None,
105        inventory: Optional[Dict] = None,
106        host_pattern: str = 'all'
107    ) -> ansible_runner.Runner:
108        """
109        Execute ad-hoc Ansible command
110
111        Args:
112            module: Ansible module name (e.g., 'shell', 'copy', 'service')
113            module_args: Module arguments
114            inventory: Inventory dictionary
115            host_pattern: Host pattern to target
116
117        Returns:
118            Runner instance with execution results
119        """
120        logger.info(f"Executing ad-hoc command: {module}")
121
122        # Prepare inventory
123        if inventory:
124            inventory_data = inventory
125        elif self.inventory:
126            inventory_data = self.inventory
127        else:
128            inventory_data = {'all': {'hosts': ['localhost']}}
129
130        # Execute ad-hoc command
131        runner = ansible_runner.run(
132            private_data_dir=str(self.private_data_dir),
133            module=module,
134            module_args=' '.join(f'{k}={v}' for k, v in (module_args or {}).items()),
135            inventory=inventory_data,
136            host_pattern=host_pattern
137        )
138
139        if runner.status == 'successful':
140            logger.info(f"✓ Command completed successfully")
141        else:
142            logger.error(f"✗ Command failed: {runner.status}")
143
144        return runner
145
146    def create_playbook(
147        self,
148        name: str,
149        tasks: List[Dict],
150        hosts: str = 'all',
151        become: bool = False,
152        vars: Optional[Dict] = None
153    ) -> Path:
154        """
155        Create Ansible playbook dynamically
156
157        Args:
158            name: Playbook name
159            tasks: List of task dictionaries
160            hosts: Target hosts
161            become: Use privilege escalation
162            vars: Playbook variables
163
164        Returns:
165            Path to created playbook
166        """
167        playbook_content = [{
168            'name': name,
169            'hosts': hosts,
170            'become': become,
171            'vars': vars or {},
172            'tasks': tasks
173        }]
174
175        playbook_file = self.private_data_dir / 'project' / f'{name}.yml'
176
177        with open(playbook_file, 'w') as f:
178            yaml.dump(playbook_content, f, default_flow_style=False)
179
180        logger.info(f"Created playbook: {playbook_file}")
181
182        return playbook_file
183
184    def get_host_facts(
185        self,
186        host: str,
187        inventory: Optional[Dict] = None
188    ) -> Dict:
189        """
190        Gather facts from target host
191
192        Args:
193            host: Target host
194            inventory: Inventory dictionary
195
196        Returns:
197            Host facts dictionary
198        """
199        logger.info(f"Gathering facts from: {host}")
200
201        runner = self.run_adhoc_command(
202            module='setup',
203            inventory=inventory,
204            host_pattern=host
205        )
206
207        # Extract facts from events
208        facts = {}
209        for event in runner.events:
210            if event.get('event') == 'runner_on_ok':
211                event_data = event.get('event_data', {})
212                if 'res' in event_data and 'ansible_facts' in event_data['res']:
213                    facts = event_data['res']['ansible_facts']
214                    break
215
216        return facts
217
218    def get_execution_summary(self) -> Dict:
219        """
220        Get summary of all playbook executions
221
222        Returns:
223            Execution summary
224        """
225        summary = {
226            'total_runs': len(self.results),
227            'successful': 0,
228            'failed': 0,
229            'results': self.results
230        }
231
232        for result in self.results:
233            if result['status'] == 'successful':
234                summary['successful'] += 1
235            else:
236                summary['failed'] += 1
237
238        return summary
239
240
241# Usage example
242runner = AnsiblePlaybookRunner(
243    private_data_dir=Path('./ansible'),
244    inventory='inventory/production.yml'
245)
246
247# Create dynamic playbook
248tasks = [
249    {
250        'name': 'Install nginx',
251        'apt': {
252            'name': 'nginx',
253            'state': 'present',
254            'update_cache': True
255        },
256        'become': True
257    },
258    {
259        'name': 'Start nginx service',
260        'service': {
261            'name': 'nginx',
262            'state': 'started',
263            'enabled': True
264        },
265        'become': True
266    },
267    {
268        'name': 'Copy nginx configuration',
269        'template': {
270            'src': 'nginx.conf.j2',
271            'dest': '/etc/nginx/nginx.conf',
272            'owner': 'root',
273            'group': 'root',
274            'mode': '0644'
275        },
276        'become': True,
277        'notify': 'Reload nginx'
278    }
279]
280
281playbook = runner.create_playbook(
282    name='deploy_nginx',
283    tasks=tasks,
284    hosts='webservers',
285    become=True,
286    vars={
287        'nginx_port': 80,
288        'nginx_user': 'www-data'
289    }
290)
291
292# Execute playbook
293result = runner.run_playbook(
294    playbook='deploy_nginx.yml',
295    inventory={
296        'webservers': {
297            'hosts': [
298                'web1.example.com',
299                'web2.example.com',
300                'web3.example.com'
301            ]
302        }
303    },
304    extravars={
305        'environment': 'production'
306    },
307    verbosity=1
308)
309
310print(f"Playbook status: {result.status}")
311print(f"Return code: {result.rc}")
312print(f"Stats: {json.dumps(result.stats, indent=2)}")
313
314# Execute ad-hoc command
315uptime_result = runner.run_adhoc_command(
316    module='shell',
317    module_args={'cmd': 'uptime'},
318    host_pattern='webservers'
319)
320
321# Get host facts
322facts = runner.get_host_facts('web1.example.com')
323print(f"\nHost Facts:")
324print(f"  OS: {facts.get('ansible_distribution')} {facts.get('ansible_distribution_version')}")
325print(f"  CPU: {facts.get('ansible_processor_vcpus')} cores")
326print(f"  Memory: {facts.get('ansible_memtotal_mb')} MB")
327
328# Get execution summary
329summary = runner.get_execution_summary()
330print(f"\nExecution Summary:")
331print(f"  Total runs: {summary['total_runs']}")
332print(f"  Successful: {summary['successful']}")
333print(f"  Failed: {summary['failed']}")

1import threading
2import queue
3from concurrent.futures import ThreadPoolExecutor, as_completed
4
5
6class AnsibleOrchestrator:
7    """
8    Advanced Ansible orchestration with parallel execution
9    """
10
11    def __init__(self, private_data_dir: Path = Path('./ansible')):
12        """
13        Initialize orchestrator
14
15        Args:
16            private_data_dir: Ansible data directory
17        """
18        self.runner = AnsiblePlaybookRunner(private_data_dir)
19        self.executor = ThreadPoolExecutor(max_workers=10)
20
21    def run_parallel_playbooks(
22        self,
23        playbooks: List[Dict[str, Any]]
24    ) -> List[Dict]:
25        """
26        Execute multiple playbooks in parallel
27
28        Args:
29            playbooks: List of playbook configurations
30
31        Returns:
32            List of execution results
33        """
34        logger.info(f"Executing {len(playbooks)} playbooks in parallel")
35
36        futures = {}
37
38        for pb_config in playbooks:
39            future = self.executor.submit(
40                self.runner.run_playbook,
41                **pb_config
42            )
43            futures[future] = pb_config['playbook']
44
45        results = []
46
47        for future in as_completed(futures):
48            playbook_name = futures[future]
49            try:
50                result = future.result()
51                results.append({
52                    'playbook': playbook_name,
53                    'status': result.status,
54                    'rc': result.rc
55                })
56                logger.info(f"✓ {playbook_name}: {result.status}")
57            except Exception as e:
58                logger.error(f"✗ {playbook_name} failed: {e}")
59                results.append({
60                    'playbook': playbook_name,
61                    'status': 'failed',
62                    'error': str(e)
63                })
64
65        return results
66
67    def run_rolling_update(
68        self,
69        playbook: str,
70        hosts: List[str],
71        batch_size: int = 1,
72        delay: int = 5
73    ) -> List[Dict]:
74        """
75        Execute rolling update across hosts
76
77        Args:
78            playbook: Playbook to execute
79            hosts: List of target hosts
80            batch_size: Number of hosts per batch
81            delay: Delay between batches in seconds
82
83        Returns:
84            List of batch results
85        """
86        logger.info(f"Starting rolling update: {playbook}")
87        logger.info(f"  Hosts: {len(hosts)}, Batch size: {batch_size}")
88
89        results = []
90
91        # Process in batches
92        for i in range(0, len(hosts), batch_size):
93            batch = hosts[i:i + batch_size]
94            batch_num = (i // batch_size) + 1
95
96            logger.info(f"Processing batch {batch_num}: {batch}")
97
98            # Execute playbook for batch
99            result = self.runner.run_playbook(
100                playbook=playbook,
101                limit=','.join(batch)
102            )
103
104            results.append({
105                'batch': batch_num,
106                'hosts': batch,
107                'status': result.status,
108                'rc': result.rc
109            })
110
111            # Delay between batches
112            if i + batch_size < len(hosts):
113                import time
114                logger.info(f"Waiting {delay}s before next batch...")
115                time.sleep(delay)
116
117        logger.info("Rolling update completed")
118
119        return results
120
121    def validate_playbook(
122        self,
123        playbook: str,
124        inventory: Optional[Dict] = None
125    ) -> Dict:
126        """
127        Validate playbook without executing
128
129        Args:
130            playbook: Playbook to validate
131            inventory: Inventory dictionary
132
133        Returns:
134            Validation results
135        """
136        logger.info(f"Validating playbook: {playbook}")
137
138        # Use --syntax-check via ansible-playbook
139        import subprocess
140
141        cmd = [
142            'ansible-playbook',
143            '--syntax-check',
144            str(self.runner.private_data_dir / 'project' / playbook)
145        ]
146
147        try:
148            result = subprocess.run(
149                cmd,
150                capture_output=True,
151                text=True,
152                check=True
153            )
154
155            logger.info("✓ Playbook syntax valid")
156
157            return {
158                'valid': True,
159                'output': result.stdout
160            }
161
162        except subprocess.CalledProcessError as e:
163            logger.error(f"✗ Playbook syntax invalid: {e.stderr}")
164
165            return {
166                'valid': False,
167                'error': e.stderr
168            }
169
170
171# Usage
172orchestrator = AnsibleOrchestrator()
173
174# Parallel playbook execution
175parallel_playbooks = [
176    {
177        'playbook': 'deploy_app.yml',
178        'inventory': {'app_servers': {'hosts': ['app1', 'app2']}},
179        'extravars': {'version': '2.0'}
180    },
181    {
182        'playbook': 'deploy_db.yml',
183        'inventory': {'db_servers': {'hosts': ['db1', 'db2']}},
184        'extravars': {'version': '1.5'}
185    },
186    {
187        'playbook': 'deploy_cache.yml',
188        'inventory': {'cache_servers': {'hosts': ['cache1', 'cache2']}},
189        'extravars': {'version': '3.0'}
190    }
191]
192
193results = orchestrator.run_parallel_playbooks(parallel_playbooks)
194
195print("\nParallel Execution Results:")
196for result in results:
197    print(f"  {result['playbook']}: {result['status']}")
198
199# Rolling update
200rolling_results = orchestrator.run_rolling_update(
201    playbook='update_system.yml',
202    hosts=['web1', 'web2', 'web3', 'web4', 'web5'],
203    batch_size=2,
204    delay=10
205)
206
207print("\nRolling Update Results:")
208for result in rolling_results:
209    print(f"  Batch {result['batch']}: {result['status']}")
210    print(f"    Hosts: {', '.join(result['hosts'])}")

1#!/usr/bin/python
2"""
3Custom Ansible Module: Cloud Resource Manager
4Manages cloud resources with custom logic
5"""
6
7from ansible.module_utils.basic import AnsibleModule
8import json
9import requests
10from typing import Dict, Any, Optional
11
12
13DOCUMENTATION = '''
14---
15module: cloud_resource
16short_description: Manage cloud resources
17description:
18    - Create, update, or delete cloud resources
19    - Supports idempotent operations
20    - Integrates with cloud provider APIs
21options:
22    name:
23        description:
24            - Resource name
25        required: true
26        type: str
27    state:
28        description:
29            - Desired state (present/absent)
30        required: false
31        default: present
32        choices: ['present', 'absent']
33        type: str
34    resource_type:
35        description:
36            - Type of resource (instance, volume, network)
37        required: true
38        type: str
39    properties:
40        description:
41            - Resource properties
42        required: false
43        type: dict
44    api_key:
45        description:
46            - API key for authentication
47        required: true
48        type: str
49author:
50    - DevOps Team
51'''
52
53EXAMPLES = '''
54# Create cloud instance
55- name: Create web server instance
56  cloud_resource:
57    name: web-server-1
58    state: present
59    resource_type: instance
60    properties:
61      size: medium
62      region: us-east-1
63      image: ubuntu-20.04
64    api_key: "{{ cloud_api_key }}"
65
66# Delete cloud instance
67- name: Remove old instance
68  cloud_resource:
69    name: old-server
70    state: absent
71    resource_type: instance
72    api_key: "{{ cloud_api_key }}"
73'''
74
75RETURN = '''
76resource:
77    description: Resource information
78    returned: always
79    type: dict
80    sample: {
81        "id": "res-12345",
82        "name": "web-server-1",
83        "state": "running",
84        "properties": {}
85    }
86changed:
87    description: Whether the resource was changed
88    returned: always
89    type: bool
90message:
91    description: Operation message
92    returned: always
93    type: str
94'''
95
96
97class CloudResourceManager:
98    """
99    Manage cloud resources via API
100    """
101
102    def __init__(
103        self,
104        api_key: str,
105        base_url: str = 'https://api.cloudprovider.com'
106    ):
107        """
108        Initialize resource manager
109
110        Args:
111            api_key: API authentication key
112            base_url: API base URL
113        """
114        self.api_key = api_key
115        self.base_url = base_url
116        self.session = requests.Session()
117        self.session.headers.update({
118            'Authorization': f'Bearer {api_key}',
119            'Content-Type': 'application/json'
120        })
121
122    def get_resource(
123        self,
124        resource_type: str,
125        name: str
126    ) -> Optional[Dict]:
127        """
128        Get resource by name
129
130        Args:
131            resource_type: Type of resource
132            name: Resource name
133
134        Returns:
135            Resource data or None
136        """
137        try:
138            response = self.session.get(
139                f'{self.base_url}/resources/{resource_type}',
140                params={'name': name}
141            )
142
143            if response.status_code == 200:
144                data = response.json()
145                if data.get('resources'):
146                    return data['resources'][0]
147
148            return None
149
150        except Exception:
151            return None
152
153    def create_resource(
154        self,
155        resource_type: str,
156        name: str,
157        properties: Dict
158    ) -> Dict:
159        """
160        Create new resource
161
162        Args:
163            resource_type: Type of resource
164            name: Resource name
165            properties: Resource properties
166
167        Returns:
168            Created resource data
169        """
170        payload = {
171            'name': name,
172            'type': resource_type,
173            'properties': properties
174        }
175
176        response = self.session.post(
177            f'{self.base_url}/resources/{resource_type}',
178            json=payload
179        )
180
181        response.raise_for_status()
182
183        return response.json()
184
185    def update_resource(
186        self,
187        resource_type: str,
188        resource_id: str,
189        properties: Dict
190    ) -> Dict:
191        """
192        Update existing resource
193
194        Args:
195            resource_type: Type of resource
196            resource_id: Resource ID
197            properties: Updated properties
198
199        Returns:
200            Updated resource data
201        """
202        response = self.session.patch(
203            f'{self.base_url}/resources/{resource_type}/{resource_id}',
204            json={'properties': properties}
205        )
206
207        response.raise_for_status()
208
209        return response.json()
210
211    def delete_resource(
212        self,
213        resource_type: str,
214        resource_id: str
215    ) -> bool:
216        """
217        Delete resource
218
219        Args:
220            resource_type: Type of resource
221            resource_id: Resource ID
222
223        Returns:
224            True if deleted successfully
225        """
226        response = self.session.delete(
227            f'{self.base_url}/resources/{resource_type}/{resource_id}'
228        )
229
230        return response.status_code in [200, 204]
231
232
233def ensure_present(
234    manager: CloudResourceManager,
235    name: str,
236    resource_type: str,
237    properties: Dict
238) -> Dict[str, Any]:
239    """
240    Ensure resource exists with desired properties
241
242    Args:
243        manager: Resource manager instance
244        name: Resource name
245        resource_type: Type of resource
246        properties: Desired properties
247
248    Returns:
249        Result dictionary
250    """
251    # Check if resource exists
252    existing = manager.get_resource(resource_type, name)
253
254    if existing:
255        # Resource exists - check if update needed
256        current_props = existing.get('properties', {})
257
258        if current_props == properties:
259            # No changes needed
260            return {
261                'changed': False,
262                'resource': existing,
263                'message': f'Resource {name} already in desired state'
264            }
265        else:
266            # Update required
267            updated = manager.update_resource(
268                resource_type,
269                existing['id'],
270                properties
271            )
272
273            return {
274                'changed': True,
275                'resource': updated,
276                'message': f'Resource {name} updated'
277            }
278    else:
279        # Create new resource
280        created = manager.create_resource(
281            resource_type,
282            name,
283            properties
284        )
285
286        return {
287            'changed': True,
288            'resource': created,
289            'message': f'Resource {name} created'
290        }
291
292
293def ensure_absent(
294    manager: CloudResourceManager,
295    name: str,
296    resource_type: str
297) -> Dict[str, Any]:
298    """
299    Ensure resource does not exist
300
301    Args:
302        manager: Resource manager instance
303        name: Resource name
304        resource_type: Type of resource
305
306    Returns:
307        Result dictionary
308    """
309    # Check if resource exists
310    existing = manager.get_resource(resource_type, name)
311
312    if existing:
313        # Delete resource
314        deleted = manager.delete_resource(
315            resource_type,
316            existing['id']
317        )
318
319        if deleted:
320            return {
321                'changed': True,
322                'resource': {},
323                'message': f'Resource {name} deleted'
324            }
325        else:
326            return {
327                'failed': True,
328                'msg': f'Failed to delete resource {name}'
329            }
330    else:
331        # Resource doesn't exist
332        return {
333            'changed': False,
334            'resource': {},
335            'message': f'Resource {name} already absent'
336        }
337
338
339def run_module():
340    """
341    Main module execution
342    """
343    # Define module arguments
344    module_args = dict(
345        name=dict(type='str', required=True),
346        state=dict(type='str', default='present', choices=['present', 'absent']),
347        resource_type=dict(type='str', required=True),
348        properties=dict(type='dict', default={}),
349        api_key=dict(type='str', required=True, no_log=True)
350    )
351
352    # Initialize module
353    module = AnsibleModule(
354        argument_spec=module_args,
355        supports_check_mode=True
356    )
357
358    # Get parameters
359    name = module.params['name']
360    state = module.params['state']
361    resource_type = module.params['resource_type']
362    properties = module.params['properties']
363    api_key = module.params['api_key']
364
365    # Initialize resource manager
366    try:
367        manager = CloudResourceManager(api_key)
368    except Exception as e:
369        module.fail_json(msg=f'Failed to initialize manager: {str(e)}')
370
371    # Check mode
372    if module.check_mode:
373        module.exit_json(changed=False, message='Check mode - no changes made')
374
375    # Execute based on desired state
376    try:
377        if state == 'present':
378            result = ensure_present(manager, name, resource_type, properties)
379        else:
380            result = ensure_absent(manager, name, resource_type)
381
382        # Check for failures
383        if result.get('failed'):
384            module.fail_json(**result)
385
386        # Success
387        module.exit_json(**result)
388
389    except Exception as e:
390        module.fail_json(msg=f'Module execution failed: {str(e)}')
391
392
393def main():
394    """
395    Entry point
396    """
397    run_module()
398
399
400if __name__ == '__main__':
401    main()

1"""
2Example playbook using custom module
3"""
4
5playbook_content = """
6---
7- name: Manage Cloud Infrastructure
8  hosts: localhost
9  gather_facts: no
10
11  vars:
12    cloud_api_key: "{{ lookup('env', 'CLOUD_API_KEY') }}"
13
14  tasks:
15    - name: Create web server instances
16      cloud_resource:
17        name: "web-server-{{ item }}"
18        state: present
19        resource_type: instance
20        properties:
21          size: medium
22          region: us-east-1
23          image: ubuntu-20.04
24          tags:
25            role: webserver
26            environment: production
27        api_key: "{{ cloud_api_key }}"
28      loop: [1, 2, 3]
29      register: web_servers
30
31    - name: Create load balancer
32      cloud_resource:
33        name: web-lb-prod
34        state: present
35        resource_type: loadbalancer
36        properties:
37          algorithm: round_robin
38          backend_instances: "{{ web_servers.results | map(attribute='resource.id') | list }}"
39          health_check:
40            protocol: http
41            port: 80
42            path: /health
43            interval: 30
44        api_key: "{{ cloud_api_key }}"
45
46    - name: Create database
47      cloud_resource:
48        name: app-database
49        state: present
50        resource_type: database
51        properties:
52          engine: postgresql
53          version: "13"
54          size: large
55          backup_retention: 7
56        api_key: "{{ cloud_api_key }}"
57
58    - name: Remove old resources
59      cloud_resource:
60        name: "{{ item }}"
61        state: absent
62        resource_type: instance
63        api_key: "{{ cloud_api_key }}"
64      loop:
65        - old-server-1
66        - old-server-2
67"""
68
69# Save playbook
70with open('./ansible/project/manage_infrastructure.yml', 'w') as f:
71    f.write(playbook_content)
72
73# Execute with ansible-runner
74runner = AnsiblePlaybookRunner()
75
76result = runner.run_playbook(
77    playbook='manage_infrastructure.yml',
78    extravars={
79        'environment': 'production'
80    }
81)
82
83print(f"Infrastructure management: {result.status}")

1import unittest
2from unittest.mock import Mock, patch
3import sys
4import json
5
6
7class TestCloudResourceModule(unittest.TestCase):
8    """
9    Test custom Ansible module
10    """
11
12    def setUp(self):
13        """Set up test fixtures"""
14        self.mock_module = Mock()
15        self.mock_module.params = {
16            'name': 'test-resource',
17            'state': 'present',
18            'resource_type': 'instance',
19            'properties': {'size': 'small'},
20            'api_key': 'test-key'
21        }
22
23    def test_resource_creation(self):
24        """Test resource creation"""
25        manager = Mock()
26        manager.get_resource.return_value = None
27        manager.create_resource.return_value = {
28            'id': 'res-123',
29            'name': 'test-resource',
30            'properties': {'size': 'small'}
31        }
32
33        result = ensure_present(
34            manager,
35            'test-resource',
36            'instance',
37            {'size': 'small'}
38        )
39
40        self.assertTrue(result['changed'])
41        self.assertEqual(result['resource']['id'], 'res-123')
42
43    def test_resource_already_exists(self):
44        """Test idempotency - resource already in desired state"""
45        manager = Mock()
46        manager.get_resource.return_value = {
47            'id': 'res-123',
48            'name': 'test-resource',
49            'properties': {'size': 'small'}
50        }
51
52        result = ensure_present(
53            manager,
54            'test-resource',
55            'instance',
56            {'size': 'small'}
57        )
58
59        self.assertFalse(result['changed'])
60
61    def test_resource_update(self):
62        """Test resource update"""
63        manager = Mock()
64        manager.get_resource.return_value = {
65            'id': 'res-123',
66            'name': 'test-resource',
67            'properties': {'size': 'small'}
68        }
69        manager.update_resource.return_value = {
70            'id': 'res-123',
71            'name': 'test-resource',
72            'properties': {'size': 'medium'}
73        }
74
75        result = ensure_present(
76            manager,
77            'test-resource',
78            'instance',
79            {'size': 'medium'}
80        )
81
82        self.assertTrue(result['changed'])
83
84    def test_resource_deletion(self):
85        """Test resource deletion"""
86        manager = Mock()
87        manager.get_resource.return_value = {
88            'id': 'res-123',
89            'name': 'test-resource'
90        }
91        manager.delete_resource.return_value = True
92
93        result = ensure_absent(
94            manager,
95            'test-resource',
96            'instance'
97        )
98
99        self.assertTrue(result['changed'])
100
101    def test_resource_already_absent(self):
102        """Test idempotency - resource already absent"""
103        manager = Mock()
104        manager.get_resource.return_value = None
105
106        result = ensure_absent(
107            manager,
108            'test-resource',
109            'instance'
110        )
111
112        self.assertFalse(result['changed'])
113
114
115# Run tests
116if __name__ == '__main__':
117    unittest.main()

1import yaml
2from pathlib import Path
3from typing import Dict, List, Any
4import json
5
6
7class AnsibleInventoryManager:
8    """
9    Manage Ansible inventory programmatically
10    """
11
12    def __init__(self, inventory_dir: Path = Path('./ansible/inventory')):
13        """
14        Initialize inventory manager
15
16        Args:
17            inventory_dir: Directory for inventory files
18        """
19        self.inventory_dir = Path(inventory_dir)
20        self.inventory_dir.mkdir(parents=True, exist_ok=True)
21
22        self.inventory = {
23            '_meta': {
24                'hostvars': {}
25            },
26            'all': {
27                'children': []
28            }
29        }
30
31    def add_host(
32        self,
33        hostname: str,
34        groups: List[str] = None,
35        variables: Dict[str, Any] = None
36    ):
37        """
38        Add host to inventory
39
40        Args:
41            hostname: Host name or IP
42            groups: List of groups host belongs to
43            variables: Host-specific variables
44        """
45        # Add host variables
46        if variables:
47            self.inventory['_meta']['hostvars'][hostname] = variables
48
49        # Add to groups
50        if groups:
51            for group in groups:
52                if group not in self.inventory:
53                    self.inventory[group] = {'hosts': []}
54
55                if hostname not in self.inventory[group]['hosts']:
56                    self.inventory[group]['hosts'].append(hostname)
57
58                # Add group to 'all' children
59                if group not in self.inventory['all']['children']:
60                    self.inventory['all']['children'].append(group)
61
62    def add_group(
63        self,
64        group_name: str,
65        variables: Dict[str, Any] = None,
66        children: List[str] = None
67    ):
68        """
69        Add group to inventory
70
71        Args:
72            group_name: Group name
73            variables: Group variables
74            children: Child groups
75        """
76        if group_name not in self.inventory:
77            self.inventory[group_name] = {}
78
79        if variables:
80            self.inventory[group_name]['vars'] = variables
81
82        if children:
83            self.inventory[group_name]['children'] = children
84
85            # Add to 'all' children
86            if group_name not in self.inventory['all']['children']:
87                self.inventory['all']['children'].append(group_name)
88
89    def set_group_vars(
90        self,
91        group_name: str,
92        variables: Dict[str, Any]
93    ):
94        """
95        Set variables for a group
96
97        Args:
98            group_name: Group name
99            variables: Variables to set
100        """
101        if group_name not in self.inventory:
102            self.inventory[group_name] = {}
103
104        if 'vars' not in self.inventory[group_name]:
105            self.inventory[group_name]['vars'] = {}
106
107        self.inventory[group_name]['vars'].update(variables)
108
109    def generate_from_cloud(
110        self,
111        cloud_provider: str,
112        region: str = None
113    ):
114        """
115        Generate inventory from cloud provider
116
117        Args:
118            cloud_provider: Cloud provider (aws, azure, gcp)
119            region: Cloud region
120        """
121        # Example: AWS EC2 instance discovery
122        if cloud_provider == 'aws':
123            import boto3
124
125            ec2 = boto3.client('ec2', region_name=region or 'us-east-1')
126            response = ec2.describe_instances()
127
128            for reservation in response['Reservations']:
129                for instance in reservation['Instances']:
130                    if instance['State']['Name'] != 'running':
131                        continue
132
133                    # Get instance details
134                    instance_id = instance['InstanceId']
135                    private_ip = instance.get('PrivateIpAddress', instance_id)
136                    tags = {tag['Key']: tag['Value'] for tag in instance.get('Tags', [])}
137
138                    # Determine groups from tags
139                    groups = []
140
141                    if 'Role' in tags:
142                        groups.append(tags['Role'])
143
144                    if 'Environment' in tags:
145                        groups.append(tags['Environment'])
146
147                    # Host variables
148                    host_vars = {
149                        'ansible_host': private_ip,
150                        'instance_id': instance_id,
151                        'instance_type': instance['InstanceType'],
152                        'availability_zone': instance['Placement']['AvailabilityZone']
153                    }
154
155                    # Add to inventory
156                    self.add_host(
157                        hostname=tags.get('Name', instance_id),
158                        groups=groups or ['ungrouped'],
159                        variables=host_vars
160                    )
161
162    def save_yaml(self, filename: str = 'inventory.yml'):
163        """
164        Save inventory as YAML file
165
166        Args:
167            filename: Output filename
168        """
169        filepath = self.inventory_dir / filename
170
171        with open(filepath, 'w') as f:
172            yaml.dump(self.inventory, f, default_flow_style=False)
173
174        logger.info(f"Inventory saved: {filepath}")
175
176    def save_json(self, filename: str = 'inventory.json'):
177        """
178        Save inventory as JSON file
179
180        Args:
181            filename: Output filename
182        """
183        filepath = self.inventory_dir / filename
184
185        with open(filepath, 'w') as f:
186            json.dump(self.inventory, f, indent=2)
187
188        logger.info(f"Inventory saved: {filepath}")
189
190    def get_inventory(self) -> Dict:
191        """
192        Get inventory dictionary
193
194        Returns:
195            Inventory dictionary
196        """
197        return self.inventory
198
199
200# Usage example
201inventory = AnsibleInventoryManager()
202
203# Add web servers
204for i in range(1, 4):
205    inventory.add_host(
206        hostname=f'web{i}.example.com',
207        groups=['webservers', 'production'],
208        variables={
209            'ansible_host': f'10.0.1.{i}',
210            'ansible_user': 'ubuntu',
211            'nginx_port': 80
212        }
213    )
214
215# Add database servers
216for i in range(1, 3):
217    inventory.add_host(
218        hostname=f'db{i}.example.com',
219        groups=['databases', 'production'],
220        variables={
221            'ansible_host': f'10.0.2.{i}',
222            'ansible_user': 'ubuntu',
223            'postgresql_port': 5432
224        }
225    )
226
227# Add cache servers
228inventory.add_host(
229    hostname='cache1.example.com',
230    groups=['cache', 'production'],
231    variables={
232        'ansible_host': '10.0.3.1',
233        'ansible_user': 'ubuntu',
234        'redis_port': 6379
235    }
236)
237
238# Set group variables
239inventory.set_group_vars('webservers', {
240    'http_port': 80,
241    'max_clients': 200
242})
243
244inventory.set_group_vars('databases', {
245    'db_port': 5432,
246    'max_connections': 100
247})
248
249inventory.set_group_vars('production', {
250    'environment': 'production',
251    'monitoring_enabled': True
252})
253
254# Save inventory
255inventory.save_yaml('production.yml')
256inventory.save_json('production.json')
257
258print("Inventory created successfully!")
259print(f"Total hosts: {len(inventory.inventory['_meta']['hostvars'])}")