Coverage for mcpgateway/tools/builder/schema.py: 100%

1# -*- coding: utf-8 -*-

2"""Location: ./mcpgateway/tools/builder/schema.py

4SPDX-License-Identifier: Apache-2.0

5Authors: Teryl Taylor

7Pydantic schemas for MCP Stack configuration validation"""

9# Standard

10from typing import Any, Dict, List, Literal, Optional

12# Third-Party

13from pydantic import BaseModel, ConfigDict, Field, field_validator

16class OpenShiftConfig(BaseModel):

17 """OpenShift-specific configuration.

19 Routes are OpenShift's native way of exposing services externally (predates Kubernetes Ingress).

20 They provide built-in TLS termination and are integrated with OpenShift's router/HAProxy infrastructure.

22 Attributes:

23 create_routes: Create OpenShift Route resources for external access (default: False)

24 domain: OpenShift apps domain for route hostnames (default: auto-detected from cluster)

25 tls_termination: TLS termination mode - edge, passthrough, or reencrypt (default: edge)

27 Examples:

28 >>> # Test with default values

29 >>> config = OpenShiftConfig()

30 >>> config.create_routes

31 False

32 >>> config.tls_termination

33 'edge'

35 >>> # Test with custom values

36 >>> config = OpenShiftConfig(

37 ... create_routes=True,

38 ... domain="apps.example.com",

39 ... tls_termination="passthrough"

40 ... )

41 >>> config.create_routes

42 True

43 >>> config.domain

44 'apps.example.com'

45 >>> config.tls_termination

46 'passthrough'

48 >>> # Test valid TLS termination modes

49 >>> for mode in ["edge", "passthrough", "reencrypt"]:

50 ... cfg = OpenShiftConfig(tls_termination=mode)

51 ... cfg.tls_termination == mode

52 True

53 True

54 True

55 """

57 create_routes: bool = Field(False, description="Create OpenShift Route resources")

58 domain: Optional[str] = Field(None, description="OpenShift apps domain (e.g., apps-crc.testing)")

59 tls_termination: Literal["edge", "passthrough", "reencrypt"] = Field("edge", description="TLS termination mode")

62class DeploymentConfig(BaseModel):

63 """Deployment configuration

65 Examples:

66 >>> # Test compose deployment

67 >>> config = DeploymentConfig(type="compose", project_name="test-project")

68 >>> config.type

69 'compose'

70 >>> config.project_name

71 'test-project'

73 >>> # Test kubernetes deployment

74 >>> config = DeploymentConfig(type="kubernetes", namespace="mcp-test")

75 >>> config.type

76 'kubernetes'

77 >>> config.namespace

78 'mcp-test'

80 >>> # Test container engine options

81 >>> config = DeploymentConfig(type="compose", container_engine="podman")

82 >>> config.container_engine

83 'podman'

85 >>> # Test with OpenShift config

86 >>> config = DeploymentConfig(

87 ... type="kubernetes",

88 ... namespace="test",

89 ... openshift=OpenShiftConfig(create_routes=True)

90 ... )

91 >>> config.openshift.create_routes

92 True

93 """

95 type: Literal["kubernetes", "compose"] = Field(..., description="Deployment type")

96 container_engine: Optional[str] = Field(default=None, description="Container engine: 'podman', 'docker', or full path (e.g., '/opt/podman/bin/podman')")

97 project_name: Optional[str] = Field(None, description="Project name for compose")

98 namespace: Optional[str] = Field(None, description="Namespace for Kubernetes")

99 openshift: Optional[OpenShiftConfig] = Field(None, description="OpenShift-specific configuration")

100

101

102class RegistryConfig(BaseModel):

103 """Container registry configuration.

104

105 Optional configuration for pushing built images to a container registry.

106 When enabled, images will be tagged with the full registry path and optionally pushed.

107

108 Authentication:

109 Users must authenticate to the registry before running the build:

110 - Docker Hub: `docker login`

111 - Quay.io: `podman login quay.io`

112 - OpenShift internal: `podman login $(oc registry info) -u $(oc whoami) -p $(oc whoami -t)`

113 - Private registry: `podman login your-registry.com -u username`

114

115 Attributes:

116 enabled: Enable registry integration (default: False)

117 url: Registry URL (e.g., "docker.io", "quay.io", "default-route-openshift-image-registry.apps-crc.testing")

118 namespace: Registry namespace/organization/project (e.g., "myorg", "mcp-gateway-test")

119 push: Push image after build (default: True)

120 image_pull_policy: Kubernetes imagePullPolicy (default: "IfNotPresent")

121

122 Examples:

123 >>> # Test with defaults (registry disabled)

124 >>> config = RegistryConfig()

125 >>> config.enabled

126 False

127 >>> config.push

128 True

129 >>> config.image_pull_policy

130 'IfNotPresent'

131

132 >>> # Test Docker Hub configuration

133 >>> config = RegistryConfig(

134 ... enabled=True,

135 ... url="docker.io",

136 ... namespace="myusername"

137 ... )

138 >>> config.enabled

139 True

140 >>> config.url

141 'docker.io'

142 >>> config.namespace

143 'myusername'

144

145 >>> # Test with custom pull policy

146 >>> config = RegistryConfig(

147 ... enabled=True,

148 ... url="quay.io",

149 ... namespace="myorg",

150 ... image_pull_policy="Always"

151 ... )

152 >>> config.image_pull_policy

153 'Always'

154

155 >>> # Test tag-only mode (no push)

156 >>> config = RegistryConfig(

157 ... enabled=True,

158 ... url="registry.local",

159 ... namespace="test",

160 ... push=False

161 ... )

162 >>> config.push

163 False

164 """

165

166 enabled: bool = Field(False, description="Enable registry push")

167 url: Optional[str] = Field(None, description="Registry URL (e.g., docker.io, quay.io, or internal registry)")

168 namespace: Optional[str] = Field(None, description="Registry namespace/organization/project")

169 push: bool = Field(True, description="Push image after build")

170 image_pull_policy: Optional[str] = Field("IfNotPresent", description="Kubernetes imagePullPolicy (IfNotPresent, Always, Never)")

171

172

173class BuildableConfig(BaseModel):

174 """Base class for components that can be built from source or use pre-built images.

175

176 This base class provides common configuration for both gateway and plugins,

177 supporting two build modes:

178 1. Pre-built image: Specify only 'image' field

179 2. Build from source: Specify 'repo' and optionally 'ref', 'context', 'containerfile', 'target'

180

181 Attributes:

182 image: Pre-built Docker image name (e.g., "mcpgateway/mcpgateway:latest")

183 repo: Git repository URL to build from

184 ref: Git branch/tag/commit to checkout (default: "main")

185 context: Build context subdirectory within repo (default: ".")

186 containerfile: Path to Containerfile/Dockerfile (default: "Containerfile")

187 target: Target stage for multi-stage builds (optional)

188 host_port: Host port mapping for direct access (optional)

189 env_vars: Environment variables for container

190 env_file: Path to environment file (.env)

191 mtls_enabled: Enable mutual TLS authentication (default: True)

192 """

193

194 # Allow attribute assignment after model creation (needed for auto-detection of env_file)

195 model_config = ConfigDict(validate_assignment=True)

196

197 # Build configuration

198 image: Optional[str] = Field(None, description="Pre-built Docker image")

199 repo: Optional[str] = Field(None, description="Git repository URL")

200 ref: Optional[str] = Field("main", description="Git branch/tag/commit")

201 context: Optional[str] = Field(".", description="Build context subdirectory")

202 containerfile: Optional[str] = Field("Containerfile", description="Containerfile path")

203 target: Optional[str] = Field(None, description="Multi-stage build target")

204

205 # Runtime configuration

206 host_port: Optional[int] = Field(None, description="Host port mapping")

207 env_vars: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Environment variables")

208 env_file: Optional[str] = Field(None, description="Path to environment file (.env)")

209 mtls_enabled: Optional[bool] = Field(True, description="Enable mTLS")

210

211 # Registry configuration

212 registry: Optional[RegistryConfig] = Field(None, description="Container registry configuration")

213

214 def model_post_init(self, _: Any) -> None:

215 """Validate that either image or repo is specified

216

217 Raises:

218 ValueError: If neither image nor repo is specified

219

220 Examples:

221 >>> # Test that error is raised when neither image nor repo specified

222 >>> try:

223 ... # BuildableConfig can't be instantiated directly, use GatewayConfig

224 ... from mcpgateway.tools.builder.schema import GatewayConfig

225 ... GatewayConfig()

226 ... except ValueError as e:

227 ... "must specify either 'image' or 'repo'" in str(e)

228 True

229

230 >>> # Test valid config with image

231 >>> from mcpgateway.tools.builder.schema import GatewayConfig

232 >>> config = GatewayConfig(image="mcpgateway:latest")

233 >>> config.image

234 'mcpgateway:latest'

235

236 >>> # Test valid config with repo

237 >>> from mcpgateway.tools.builder.schema import GatewayConfig

238 >>> config = GatewayConfig(repo="https://github.com/example/repo")

239 >>> config.repo

240 'https://github.com/example/repo'

241 """

242 if not self.image and not self.repo:

243 component_type = self.__class__.__name__.replace("Config", "")

244 raise ValueError(f"{component_type} must specify either 'image' or 'repo'")

245

246

247class GatewayConfig(BuildableConfig):

248 """Gateway configuration.

249

250 Extends BuildableConfig to support either pre-built gateway images or

251 building the gateway from source repository.

252

253 Attributes:

254 port: Gateway internal port (default: 4444)

255

256 Examples:

257 >>> # Test with pre-built image

258 >>> config = GatewayConfig(image="mcpgateway:latest")

259 >>> config.image

260 'mcpgateway:latest'

261 >>> config.port

262 4444

263

264 >>> # Test with custom port

265 >>> config = GatewayConfig(image="mcpgateway:latest", port=8080)

266 >>> config.port

267 8080

268

269 >>> # Test with source repository

270 >>> config = GatewayConfig(

271 ... repo="https://github.com/example/gateway",

272 ... ref="v1.0.0"

273 ... )

274 >>> config.repo

275 'https://github.com/example/gateway'

276 >>> config.ref

277 'v1.0.0'

278

279 >>> # Test with environment variables

280 >>> config = GatewayConfig(

281 ... image="mcpgateway:latest",

282 ... env_vars={"LOG_LEVEL": "DEBUG", "PORT": "4444"}

283 ... )

284 >>> config.env_vars['LOG_LEVEL']

285 'DEBUG'

286

287 >>> # Test with mTLS enabled

288 >>> config = GatewayConfig(image="mcpgateway:latest", mtls_enabled=True)

289 >>> config.mtls_enabled

290 True

291 """

292

293 port: Optional[int] = Field(4444, description="Gateway port")

294

295

296class PluginConfig(BuildableConfig):

297 """Plugin configuration.

298

299 Extends BuildableConfig to support plugin-specific configuration while

300 inheriting common build and runtime capabilities.

301

302 Attributes:

303 name: Unique plugin identifier

304 port: Plugin internal port (default: 8000)

305 expose_port: Whether to expose plugin port on host (default: False)

306 plugin_overrides: Plugin-specific override configuration

307 """

308

309 name: str = Field(..., description="Plugin name")

310 port: Optional[int] = Field(8000, description="Plugin port")

311 expose_port: Optional[bool] = Field(False, description="Expose port on host")

312 plugin_overrides: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Plugin overrides")

313

314 @field_validator("name")

315 @classmethod

316 def validate_name(cls, v: str) -> str:

317 """Validate plugin name is non-empty

318

319 Args:

320 v: Plugin name value to validate

321

322 Returns:

323 Validated plugin name

324

325 Raises:

326 ValueError: If plugin name is empty or whitespace only

327

328 Examples:

329 >>> # Test valid plugin names

330 >>> PluginConfig.validate_name("my-plugin")

331 'my-plugin'

332 >>> PluginConfig.validate_name("plugin_123")

333 'plugin_123'

334 >>> PluginConfig.validate_name("TestPlugin")

335 'TestPlugin'

336

337 >>> # Test empty name raises error

338 >>> try:

339 ... PluginConfig.validate_name("")

340 ... except ValueError as e:

341 ... "cannot be empty" in str(e)

342 True

343

344 >>> # Test whitespace-only name raises error

345 >>> try:

346 ... PluginConfig.validate_name(" ")

347 ... except ValueError as e:

348 ... "cannot be empty" in str(e)

349 True

350 """

351 if not v or not v.strip():

352 raise ValueError("Plugin name cannot be empty")

353 return v

354

355

356class CertificatesConfig(BaseModel):

357 """Certificate configuration.

358

359 Supports two modes:

360 1. Local certificate generation (use_cert_manager=false, default):

361 - Certificates generated locally using OpenSSL (via Makefile)

362 - Deployed to Kubernetes as secrets via kubectl

363 - Manual rotation required before expiry

364

365 2. cert-manager integration (use_cert_manager=true, Kubernetes only):

366 - Certificates managed by cert-manager controller

367 - Automatic renewal before expiry (default: at 2/3 of lifetime)

368 - Native Kubernetes Certificate resources

369 - Requires cert-manager to be installed in cluster

370

371 Attributes:

372 validity_days: Certificate validity period in days (default: 825 ≈ 2.25 years)

373 auto_generate: Auto-generate certificates locally (default: True)

374 use_cert_manager: Use cert-manager for certificate management (default: False, Kubernetes only)

375 cert_manager_issuer: Name of cert-manager Issuer/ClusterIssuer (default: "mcp-ca-issuer")

376 cert_manager_kind: Type of issuer - Issuer or ClusterIssuer (default: "Issuer")

377 ca_path: Path to CA certificates for local generation (default: "./certs/mcp/ca")

378 gateway_path: Path to gateway certificates for local generation (default: "./certs/mcp/gateway")

379 plugins_path: Path to plugin certificates for local generation (default: "./certs/mcp/plugins")

380 """

381

382 validity_days: Optional[int] = Field(825, description="Certificate validity in days")

383 auto_generate: Optional[bool] = Field(True, description="Auto-generate certificates locally")

384

385 # cert-manager integration (Kubernetes only)

386 use_cert_manager: Optional[bool] = Field(False, description="Use cert-manager for certificate management (Kubernetes only)")

387 cert_manager_issuer: Optional[str] = Field("mcp-ca-issuer", description="cert-manager Issuer/ClusterIssuer name")

388 cert_manager_kind: Optional[Literal["Issuer", "ClusterIssuer"]] = Field("Issuer", description="cert-manager issuer kind")

389

390 ca_path: Optional[str] = Field("./certs/mcp/ca", description="CA certificate path")

391 gateway_path: Optional[str] = Field("./certs/mcp/gateway", description="Gateway cert path")

392 plugins_path: Optional[str] = Field("./certs/mcp/plugins", description="Plugins cert path")

393

394

395class PostgresConfig(BaseModel):

396 """PostgreSQL database configuration"""

397

398 enabled: Optional[bool] = Field(True, description="Enable PostgreSQL deployment")

399 image: Optional[str] = Field("quay.io/sclorg/postgresql-15-c9s:latest", description="PostgreSQL image (default is OpenShift-compatible)")

400 database: Optional[str] = Field("mcp", description="Database name")

401 user: Optional[str] = Field("postgres", description="Database user")

402 password: Optional[str] = Field("mysecretpassword", description="Database password")

403 storage_size: Optional[str] = Field("10Gi", description="Persistent volume size (Kubernetes only)")

404 storage_class: Optional[str] = Field(None, description="Storage class name (Kubernetes only)")

405

406

407class RedisConfig(BaseModel):

408 """Redis cache configuration"""

409

410 enabled: Optional[bool] = Field(True, description="Enable Redis deployment")

411 image: Optional[str] = Field("redis:latest", description="Redis image")

412

413

414class InfrastructureConfig(BaseModel):

415 """Infrastructure services configuration"""

416

417 postgres: Optional[PostgresConfig] = Field(default_factory=PostgresConfig)

418 redis: Optional[RedisConfig] = Field(default_factory=RedisConfig)

419

420

421class MCPStackConfig(BaseModel):

422 """Complete MCP Stack configuration"""

423

424 deployment: DeploymentConfig

425 gateway: GatewayConfig

426 plugins: List[PluginConfig] = Field(default_factory=list)

427 certificates: Optional[CertificatesConfig] = Field(default_factory=CertificatesConfig)

428 infrastructure: Optional[InfrastructureConfig] = Field(default_factory=InfrastructureConfig)

429

430 @field_validator("plugins")

431 @classmethod

432 def validate_plugin_names_unique(cls, v: List[PluginConfig]) -> List[PluginConfig]:

433 """Ensure plugin names are unique

434

435 Args:

436 v: List of plugin configurations to validate

437

438 Returns:

439 Validated list of plugin configurations

440

441 Raises:

442 ValueError: If duplicate plugin names are found

443

444 Examples:

445 >>> from mcpgateway.tools.builder.schema import PluginConfig

446 >>> # Test with unique names (valid)

447 >>> plugins = [

448 ... PluginConfig(name="plugin1", image="img1:latest"),

449 ... PluginConfig(name="plugin2", image="img2:latest")

450 ... ]

451 >>> result = MCPStackConfig.validate_plugin_names_unique(plugins)

452 >>> len(result) == 2

453 True

454

455 >>> # Test with duplicate names (invalid)

456 >>> try:

457 ... duplicates = [

458 ... PluginConfig(name="duplicate", image="img1:latest"),

459 ... PluginConfig(name="duplicate", image="img2:latest")

460 ... ]

461 ... MCPStackConfig.validate_plugin_names_unique(duplicates)

462 ... except ValueError as e:

463 ... "Duplicate plugin names found" in str(e)

464 True

465

466 >>> # Test with empty list (valid)

467 >>> empty = MCPStackConfig.validate_plugin_names_unique([])

468 >>> len(empty) == 0

469 True

470 """

471 names = [p.name for p in v]

472 if len(names) != len(set(names)):

473 duplicates = [name for name in names if names.count(name) > 1]

474 raise ValueError(f"Duplicate plugin names found: {duplicates}")

475 return v

Coverage for mcpgateway / tools / builder / schema.py: 100%

85 statements