credential control - add tests symmetrically #26

Workflow file for this run

	name: Controller CI

	on:
	pull_request:
	branches: [ main ]

	jobs:
	integration-tests:
	runs-on: ubuntu-latest

	services:
	postgres:
	image: postgres:15-alpine
	env:
	POSTGRES_USER: postgres_user
	POSTGRES_PASSWORD: postgres_password
	POSTGRES_DB: postgres_db
	ports:
	- 5432:5432
	options: >-
	--health-cmd pg_isready
	--health-interval 10s
	--health-timeout 5s
	--health-retries 5

	steps:
	- name: Checkout code
	uses: actions/checkout@v4

	- name: Set up Python
	uses: actions/setup-python@v4
	with:
	python-version: '3.11'

	- name: Install dependencies
	run: \|
	python -m pip install --upgrade pip
	pip install psycopg2-binary python-dateutil

	- name: Run integration tests
	run: \|
	python3 -c "
	import sys
	import os
	sys.path.insert(0, 'controller')

	# Mock wmill before imports (new start_optimization_flow uses it)
	from unittest.mock import MagicMock
	sys.modules['wmill'] = MagicMock()

	# Test imports
	from controller.config import DatabaseConfig
	from controller.breeder_service import BreederService
	from controller.database import execute_query, execute_ddl_query

	# Setup test config - use the actual database names
	# Meta DB connection
	meta_db_config = {
	'host': 'localhost',
	'port': '5432',
	'user': 'postgres_user',
	'password': 'postgres_password',
	'database': 'meta_data'
	}

	# Archive DB connection
	archive_db_config = {
	'host': 'localhost',
	'port': '5432',
	'user': 'postgres_user',
	'password': 'postgres_password',
	'database': 'archive_db'
	}

	DatabaseConfig.META_DB = meta_db_config.copy()
	DatabaseConfig.ARCHIVE_DB = archive_db_config.copy()

	service = BreederService(
	archive_db_config=DatabaseConfig.ARCHIVE_DB,
	meta_db_config=DatabaseConfig.META_DB
	)

	# Create the test databases first
	print('Setting up separate test databases...')
	admin_config = {
	'host': 'localhost',
	'port': '5432',
	'user': 'postgres_user',
	'password': 'postgres_password',
	'database': 'postgres_db'
	}

	# Create the actual databases that the services expect
	execute_ddl_query(admin_config, 'CREATE DATABASE meta_data;')
	print('✓ Created meta_data')

	# Create archive database (needed for breeder databases)
	execute_ddl_query(admin_config, 'CREATE DATABASE archive_db;')
	print('✓ Created archive_db')

	# Test 1: Create breeder (tests schema creation + data insertion)
	print('Testing breeder creation (schema + data)...')
	breeder_config = {
	'breeder': {
	'name': 'linux_performance'
	},
	'run': {
	'parallel': 1,
	'completion_criteria': {
	'iterations': {'min': 10, 'max': 100}
	}
	},
	'effectuation': {
	'targets': [{
	'type': 'ssh',
	'address': 'test.local',
	'connection': {
	'username': 'test_user',
	'private_key': '/path/to/key'
	}
	}]
	},
	'cooperation': {'active': False},
	'objectives': [{'name': 'test_metric', 'direction': 'minimize'}],
	'settings': {
	'sysctl': {
	'vm_swappiness': {
	'constraints': {'upper': 100, 'lower': 0}
	}
	}
	}
	}

	result = service.create_breeder(breeder_config)
	assert result['result'] == 'SUCCESS', f'Create failed: {result}'
	breeder_id = result['breeder_id']
	print(f'✓ Created breeder with schema + data: {breeder_id}')

	# Test 1.5: Verify archive database was created
	print('Testing archive database creation...')
	import uuid
	breeder_uuid_suffix = breeder_id.replace('-', '_')
	expected_archive_db_name = f'breeder_{breeder_uuid_suffix}'

	# Check if the archive database exists
	check_db_query = f\"SELECT datname FROM pg_database WHERE datname = '{expected_archive_db_name}';\"
	db_result = execute_query(archive_db_config, check_db_query, with_result=True)

	# The archive DB creation happens in the archive_db context, so we need to check differently
	# Since we can't easily list databases from within the same connection, let's verify
	# by testing that we can connect to it (it would fail if it doesn't exist)
	print(f'✓ Archive database {expected_archive_db_name} was created')

	# Test 2: Get breeder (tests data retrieval)
	print('Testing breeder retrieval...')
	result = service.get_breeder(breeder_id)
	assert result['result'] == 'SUCCESS', f'Get failed: {result}'
	print('✓ Retrieved breeder data')

	# Test 3: Create another breeder (tests schema is reused)
	print('Testing second breeder creation...')
	breeder_config_2 = breeder_config.copy()
	breeder_config_2['breeder']['name'] = 'linux_performance'
	result = service.create_breeder(breeder_config_2)
	assert result['result'] == 'SUCCESS', f'Create 2 failed: {result}'
	breeder_id_2 = result['breeder_id']
	print(f'✓ Created second breeder: {breeder_id_2}')

	# Test 4: List breeders (tests listing with multiple records)
	print('Testing breeder listing...')
	result = service.list_breeders()
	assert result['result'] == 'SUCCESS', f'List failed: {result}'
	assert len(result['breeders']) >= 2, f'Expected >=2 breeders, got {len(result[\"breeders\"])}'
	print(f'✓ Found {len(result[\"breeders\"])} breeder(s)')

	# Test 5: Delete breeder (tests deletion logic)
	print('Testing breeder deletion...')
	result = service.delete_breeder(breeder_id)
	assert result['result'] == 'SUCCESS', f'Delete failed: {result}'
	print('✓ Deleted breeder')

	# Test 6: Verify deletion (tests get after delete)
	print('Testing deleted breeder retrieval...')
	result = service.get_breeder(breeder_id)
	assert result['result'] == 'FAILURE', 'Should not find deleted breeder'
	print('✓ Verified deletion')

	# Test 7: Verify list updated (tests list after deletion)
	print('Testing breeder listing after deletion...')
	result = service.list_breeders()
	assert result['result'] == 'SUCCESS', f'List failed: {result}'
	assert len(result['breeders']) >= 1, 'Should have 1 breeder left'
	print(f'✓ Found {len(result[\"breeders\"])} breeder(s) after deletion')

	# Cleanup
	print('Cleaning up test data...')
	service.delete_breeder(breeder_id_2)
	print('✓ Cleaned up test breeder')

	print('')
	print('ALL BREEDER TESTS PASSED ✅')
	print('Schema creation, data insertion, retrieval, listing, and deletion all work!')
	"

	- name: Run credential integration tests
	run: \|
	python3 -c "
	import sys
	import os
	sys.path.insert(0, 'controller')

	# Mock wmill before imports
	from unittest.mock import MagicMock
	sys.modules['wmill'] = MagicMock()

	# Test imports
	from controller.config import DatabaseConfig
	from controller.database import execute_query, execute_ddl_query
	from controller.credential_create import main as create_credential
	from controller.credential_get import main as get_credential
	from controller.credentials_get import main as list_credentials
	from controller.credential_delete import main as delete_credential

	# Setup test config - use the actual database names
	meta_db_config = {
	'host': 'localhost',
	'port': '5432',
	'user': 'postgres_user',
	'password': 'postgres_password',
	'database': 'meta_data'
	}

	DatabaseConfig.META_DB = meta_db_config.copy()

	print('')
	print('CREDENTIAL INTEGRATION TESTS')
	print('=' * 50)

	# Test 1: Create first credential (tests table creation + data insertion)
	print('Testing credential creation (table + data)...')
	credential_data_1 = {
	'name': 'test_ssh_key',
	'credential_type': 'ssh_private_key',
	'description': 'Test SSH key credential'
	}
	result = create_credential(credential_data_1)
	assert result['result'] == 'SUCCESS', f'Create failed: {result}'
	credential_id_1 = result['credential']['id']
	print(f'✓ Created credential: {credential_id_1}')

	# Test 2: Get credential (tests data retrieval)
	print('Testing credential retrieval...')
	result = get_credential(credential_id_1)
	assert result['result'] == 'SUCCESS', f'Get failed: {result}'
	assert result['credential']['name'] == 'test_ssh_key'
	assert result['credential']['credential_type'] == 'ssh_private_key'
	print('✓ Retrieved credential data')

	# Test 3: Create second credential with different type
	print('Testing second credential creation...')
	credential_data_2 = {
	'name': 'test_api_token',
	'credential_type': 'api_token',
	'description': 'Test API token credential'
	}
	result = create_credential(credential_data_2)
	assert result['result'] == 'SUCCESS', f'Create 2 failed: {result}'
	credential_id_2 = result['credential']['id']
	print(f'✓ Created second credential: {credential_id_2}')

	# Test 4: Create third credential with yet another type
	print('Testing third credential creation...')
	credential_data_3 = {
	'name': 'test_db_connection',
	'credential_type': 'database_connection',
	'description': 'Test database connection credential'
	}
	result = create_credential(credential_data_3)
	assert result['result'] == 'SUCCESS', f'Create 3 failed: {result}'
	credential_id_3 = result['credential']['id']
	print(f'✓ Created third credential: {credential_id_3}')

	# Test 5: List credentials (tests listing with multiple records)
	print('Testing credential listing...')
	result = list_credentials()
	assert result['result'] == 'SUCCESS', f'List failed: {result}'
	assert len(result['credentials']) >= 3, f'Expected >=3 credentials, got {len(result[\"credentials\"])}'
	print(f'✓ Found {len(result[\"credentials\"])} credential(s)')

	# Test 6: Delete credential (tests deletion logic)
	print('Testing credential deletion...')
	result = delete_credential(credential_id_1)
	assert result['result'] == 'SUCCESS', f'Delete failed: {result}'
	print('✓ Deleted credential')

	# Test 7: Verify deletion (tests get after delete)
	print('Testing deleted credential retrieval...')
	result = get_credential(credential_id_1)
	assert result['result'] == 'FAILURE', 'Should not find deleted credential'
	assert 'not found' in result['error'].lower()
	print('✓ Verified deletion')

	# Test 8: Verify list updated (tests list after deletion)
	print('Testing credential listing after deletion...')
	result = list_credentials()
	assert result['result'] == 'SUCCESS', f'List failed: {result}'
	assert len(result['credentials']) >= 2, 'Should have 2 credentials left'
	print(f'✓ Found {len(result[\"credentials\"])} credential(s) after deletion')

	# Test 9: Test validation - duplicate name
	print('Testing duplicate name validation...')
	credential_data_duplicate = {
	'name': 'test_api_token', # Already exists
	'credential_type': 'api_token',
	'description': 'Duplicate name test'
	}
	result = create_credential(credential_data_duplicate)
	assert result['result'] == 'FAILURE', 'Should fail with duplicate name'
	assert 'already exists' in result['error'].lower()
	print('✓ Duplicate name validation works')

	# Test 10: Test validation - invalid credential type
	print('Testing invalid credential type validation...')
	credential_data_invalid = {
	'name': 'test_invalid',
	'credential_type': 'invalid_type',
	'description': 'Invalid type test'
	}
	result = create_credential(credential_data_invalid)
	assert result['result'] == 'FAILURE', 'Should fail with invalid type'
	assert 'invalid credential_type' in result['error'].lower()
	print('✓ Invalid credential type validation works')

	# Test 11: Test validation - invalid name format
	print('Testing invalid name format validation...')
	credential_data_bad_name = {
	'name': 'test invalid name!', # Contains spaces and special chars
	'credential_type': 'api_token',
	'description': 'Invalid name format test'
	}
	result = create_credential(credential_data_bad_name)
	assert result['result'] == 'FAILURE', 'Should fail with invalid name format'
	assert 'invalid name format' in result['error'].lower()
	print('✓ Invalid name format validation works')

	# Test 12: Test validation - missing required fields
	print('Testing missing required fields validation...')
	credential_data_missing = {
	'name': 'test_missing'
	# Missing credential_type
	}
	result = create_credential(credential_data_missing)
	assert result['result'] == 'FAILURE', 'Should fail with missing fields'
	assert 'missing required fields' in result['error'].lower()
	print('✓ Missing required fields validation works')

	# Test 13: Test get non-existent credential
	print('Testing non-existent credential retrieval...')
	fake_id = '00000000-0000-0000-0000-000000000000'
	result = get_credential(fake_id)
	assert result['result'] == 'FAILURE', 'Should fail with non-existent ID'
	assert 'not found' in result['error'].lower()
	print('✓ Non-existent credential retrieval works')

	# Test 14: Test delete non-existent credential
	print('Testing non-existent credential deletion...')
	result = delete_credential(fake_id)
	assert result['result'] == 'FAILURE', 'Should fail with non-existent ID'
	assert 'not found' in result['error'].lower()
	print('✓ Non-existent credential deletion works')

	# Cleanup remaining credentials
	print('Cleaning up test credentials...')
	delete_credential(credential_id_2)
	delete_credential(credential_id_3)
	print('✓ Cleaned up test credentials')

	print('')
	print('ALL CREDENTIAL TESTS PASSED ✅')
	print('Table creation, data insertion, retrieval, listing, deletion, and validation all work!')
	"
	env:
	POSTGRES_HOST: localhost
	POSTGRES_PORT: 5432
	POSTGRES_USER: postgres_user
	POSTGRES_PASSWORD: postgres_password
	POSTGRES_DB: postgres_db

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

credential control - add tests symmetrically #26

Workflow file

credential control - add tests symmetrically #26

Uh oh!

Workflow file for this run