()
| 137 | |
| 138 | |
| 139 | def main(): |
| 140 | # Check if skopeo supports --user-agent-prefix |
| 141 | if not skopeo_supports_user_agent_prefix(): |
| 142 | # Get skopeo version for the skip message |
| 143 | try: |
| 144 | result = subprocess.run( |
| 145 | ["skopeo", "--version"], |
| 146 | capture_output=True, |
| 147 | text=True, |
| 148 | timeout=10, |
| 149 | ) |
| 150 | version = result.stdout.strip() |
| 151 | except Exception: |
| 152 | version = "unknown" |
| 153 | |
| 154 | # On distros that should have new enough skopeo, fail hard |
| 155 | if distro_requires_user_agent_support(): |
| 156 | print(f"ERROR: skopeo ({version}) does not support --user-agent-prefix", flush=True) |
| 157 | print("This distro should have skopeo >= 1.21.0", flush=True) |
| 158 | return 1 |
| 159 | |
| 160 | print(f"SKIP: skopeo ({version}) does not support --user-agent-prefix", flush=True) |
| 161 | print("This feature requires skopeo >= 1.21.0", flush=True) |
| 162 | # Exit 0 to skip the test gracefully |
| 163 | return 0 |
| 164 | |
| 165 | print("=== User-Agent Header Test ===", flush=True) |
| 166 | |
| 167 | # Start server in background thread (port allocated dynamically) |
| 168 | server_thread = threading.Thread(target=run_server, daemon=True) |
| 169 | server_thread.start() |
| 170 | |
| 171 | # Wait for server to be ready and get the allocated port |
| 172 | if not server_ready.wait(timeout=5): |
| 173 | print("ERROR: Server failed to start", flush=True) |
| 174 | return 1 |
| 175 | |
| 176 | registry = f"127.0.0.1:{allocated_port}" |
| 177 | print(f"Server listening on {registry}", flush=True) |
| 178 | |
| 179 | # Configure insecure registry |
| 180 | registries_conf = f"""[[registry]] |
| 181 | location = "{registry}" |
| 182 | insecure = true |
| 183 | """ |
| 184 | conf_path = "/etc/containers/registries.conf.d/99-test-insecure.conf" |
| 185 | print(f"Writing registries config to {conf_path}", flush=True) |
| 186 | with open(conf_path, "w") as f: |
| 187 | f.write(registries_conf) |
| 188 | print(registries_conf, flush=True) |
| 189 | |
| 190 | try: |
| 191 | |
| 192 | # Test with bootc |
| 193 | print("\n=== Testing with bootc ===", flush=True) |
| 194 | result = subprocess.run( |
| 195 | ["bootc", "switch", "--transport", "registry", f"{registry}/test:latest"], |
| 196 | capture_output=True, |
no test coverage detected