title: Invalid DO API token produces Python exception backtrace
state: OPEN
author: dboreham (David Boreham)
labels:
comments: 0
assignees:
projects:
milestone:
number: 95
--
When I used an expired API token I got this output:
```
machine --config-file ~/.config/machine/config.yml list
Traceback (most recent call last):
File "/home/david/.local/bin/machine", line 10, in <module>
sys.exit(main())
^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/click/core.py", line 1157, in __call__
return self.main(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/click/core.py", line 1078, in main
rv = self.invoke(ctx)
^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/click/core.py", line 1688, in invoke
return _process_result(sub_ctx.command.invoke(sub_ctx))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/click/core.py", line 1434, in invoke
return ctx.invoke(self.callback, **ctx.params)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/click/core.py", line 783, in invoke
return __callback(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/click/decorators.py", line 33, in new_func
return f(get_current_context(), *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/machine/subcommands/list.py", line 81, in command
vms = get_vms(command_context, id, name, tag, type, region, all)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/machine/subcommands/list.py", line 36, in get_vms
vms = provider.list_vms(tag=TAG_MACHINE_SESSION_PREFIX + command_context.session_id)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/machine/providers/digitalocean.py", line 85, in list_vms
droplets = self._manager.get_all_droplets(tag_name=tag)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/digitalocean/Manager.py", line 66, in get_all_droplets
data = self.get_data("droplets/", params=params)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/.local/share/uv/tools/machine/lib/python3.12/site-packages/digitalocean/baseapi.py", line 233, in get_data
raise DataReadError(msg)
digitalocean.DataReadError: Unable to authenticate you
```
In this situation the result should be more graceful -- catch the exception and output a specific error message the user can clearly understand.The issue asks for graceful handling of auth failures instead of a Python traceback. Let me explore the relevant code.
1 import os
2
3 import click
4
5 from machine import config
6 from machine import constants
7 from machine.di import d
8 from machine.log import output
9 from machine.providers import create_provider
10 from machine.subcommands import check, create, destroy, info, list, projects, ssh_keys, domains, list_domain, types, status
11 from machine.types import CliOptions, MainCmdCtx
12 from machine.util import load_session_id
13
14 CLICK_CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
15
16
17 @click.group(context_settings=CLICK_CONTEXT_SETTINGS)
18 @click.option("--debug", is_flag=True, default=False, help="Enable debug output")
19 @click.option("--quiet", is_flag=True, default=False, help="Suppress all non-essential output")
20 @click.option("--verbose", is_flag=True, default=False, help="Enable verbose output")
21 @click.option("--dry-run", is_flag=True, default=False, help="Run but do not do anything")
22 @click.option("--config-file", metavar="<PATH>", help=f"Specify the config file (default {constants.default_config_file_path})")
23 @click.option(
24 "--session-id", metavar="<ID>", default=load_session_id, help="Override the default session ID (default: from session-id.yml)"
25 )
26 @click.pass_context
27 def main(context, debug, quiet, verbose, dry_run, config_file, session_id):
28 options = CliOptions(debug, quiet, verbose, dry_run)
29 d.opt = options
30 # Skip config loading for version subcommand since it doesn't need it
31 # and should work even when no config file exists (#25)
32 if context.invoked_subcommand == "version":
33 return
34 cfg = config.get(config_file)
35 provider = create_provider(cfg.provider_name, cfg.provider_config)
36 main_context = MainCmdCtx(cfg, session_id, provider)
37 context.obj = main_context
38
39
40 @main.command()
41 @click.pass_context
42 def version(context):
43 try:
44 version_file = os.path.join(os.path.dirname(__file__), "version.txt")
45 with open(version_file) as f:
46 version_string = f.read().strip()
47 except FileNotFoundError:
48 version_string = "dev"
49 output(version_string)
50
51
52 main.add_command(check.command, "check")
53 main.add_command(create.command, "create")
54 main.add_command(destroy.command, "destroy")
55 main.add_command(domains.command, "domains")
56 main.add_command(info.command, "info")
57 main.add_command(list.command, "list")
58 main.add_command(list_domain.command, "list-domain")
59 main.add_command(projects.command, "projects")
60 main.add_command(ssh_keys.command, "ssh-keys")
61 main.add_command(types.command, "types")
62 main.add_command(status.command, "status")
63 {
"file_path": "/home/david/projects/stirlingbridge/machine/machine/providers/digitalocean.py"
}1 import digitalocean
2
3 from machine.log import fatal_error, info
4 from machine.provider import CloudProvider, VM, SSHKey, DNSRecord
5
6
7 VALID_REGIONS = ["NYC1", "NYC3", "AMS3", "SFO2", "SFO3", "SGP1", "LON1", "FRA1", "TOR1", "BLR1", "SYD1"]
8
9 VALID_IMAGES = [
10 "almalinux-8-x64",
11 "almalinux-9-x64",
12 "centos-stream-9-x64",
13 "debian-11-x64",
14 "debian-12-x64",
15 "fedora-39-x64",
16 "fedora-40-x64",
17 "rockylinux-9-x64",
18 "rockylinux-8-x64",
19 "ubuntu-20-04-x64",
20 "ubuntu-22-04-x64",
21 "ubuntu-24-04-x64",
22 ]
23
24
25 def _droplet_to_vm(droplet) -> VM:
26 region = droplet.region
27 if isinstance(region, dict):
28 region = region.get("slug")
29 return VM(
30 id=str(droplet.id),
31 name=droplet.name,
32 tags=droplet.tags,
33 region=region,
34 ip_address=droplet.ip_address,
35 status=droplet.status,
36 )
37
38
39 class DigitalOceanProvider(CloudProvider):
40 def __init__(self, provider_config):
41 if "access-token" not in provider_config:
42 fatal_error("Required key 'access-token' not found in 'digital-ocean' section of config file")
43 self.token = provider_config["access-token"]
44 self._manager = digitalocean.Manager(token=self.token)
45
46 def create_vm(self, name, region, image, size, ssh_key_names, tags, user_data) -> VM:
47 ssh_keys = []
48 for ssh_key_name in ssh_key_names:
49 ssh_key = self._get_do_ssh_key(ssh_key_name)
50 if not ssh_key:
51 fatal_error(f"Error: SSH key '{ssh_key_name}' not found in DigitalOcean")
52 ssh_keys.append(ssh_key)
53
54 droplet = digitalocean.Droplet(
55 token=self.token,
56 name=name,
57 region=region,
58 image=image,
59 size_slug=size,
60 ssh_keys=ssh_keys,
61 tags=tags,
62 user_data=user_data,
63 backups=False,
64 )
65 try:
66 droplet.create()
67 except digitalocean.DataReadError as e:
68 fatal_error(f"DigitalOcean API error creating VM: {e}")
69 return _droplet_to_vm(droplet)
70
71 def get_vm(self, vm_id) -> VM:
72 droplet = self._manager.get_droplet(vm_id)
73 return _droplet_to_vm(droplet)
74
75 def destroy_vm(self, vm_id) -> bool:
76 try:
77 droplet = self._manager.get_droplet(vm_id)
78 except digitalocean.NotFoundError:
79 fatal_error(f"Error: machine with id {vm_id} not found")
80 result = droplet.destroy()
81 return result
82
83 def list_vms(self, tag=None) -> list:
84 if tag:
85 droplets = self._manager.get_all_droplets(tag_name=tag)
86 else:
87 droplets = self._manager.get_all_droplets()
88 return [_droplet_to_vm(d) for d in droplets]
89
90 def get_ssh_key(self, name) -> SSHKey:
91 do_key = self._get_do_ssh_key(name)
92 if not do_key:
93 return None
94 return SSHKey(
95 id=str(do_key.id),
96 name=do_key.name,
97 fingerprint=do_key.fingerprint,
98 public_key=do_key.public_key,
99 )
100
101 def list_ssh_keys(self) -> list:
102 keys = self._manager.get_all_sshkeys()
103 return [
104 SSHKey(id=str(k.id), name=k.name, fingerprint=k.fingerprint, public_key=k.public_key)
105 for k in keys
106 ]
107
108 def create_dns_record(self, zone, record_type, name, data, ttl, tag=None) -> str:
109 domain = digitalocean.Domain(token=self.token, name=zone)
110 try:
111 record = domain.create_new_domain_record(type=record_type, ttl=ttl, name=name, data=data, tag=tag)
112 except digitalocean.NotFoundError:
113 info(f"Warning: DNS zone '{zone}' not found in DigitalOcean, DNS record not set")
114 return None
115 return record
116
117 def delete_dns_record(self, zone, record_name) -> bool:
118 domain = digitalocean.Domain(token=self.token, name=zone)
119 records = domain.get_records()
120 for record in records:
121 if record.name == record_name:
122 domain.delete_domain_record(id=record.id)
123 return True
124 return False
125
126 def get_dns_records(self, zone) -> list:
127 domain = digitalocean.Domain(token=self.token, name=zone)
128 records = domain.get_records()
129 return [
130 DNSRecord(id=str(r.id), name=r.name, type=r.type, data=r.data, ttl=r.ttl)
131 for r in records
132 ]
133
134 def list_domains(self) -> list:
135 domains = self._manager.get_all_domains()
136 return [d.name for d in domains]
137
138 def list_projects(self) -> list:
139 projects = self._manager.get_all_projects()
140 return [p.name for p in projects]
141
142 def assign_to_project(self, project_name, vm_id):
143 projects = self._manager.get_all_projects()
144 project = None
145 for p in projects:
146 if p.name == project_name:
147 project = p
148 break
149 if not project:
150 fatal_error(f"Error: Project {project_name} does not exist, machine created but not assigned to project")
151 project.assign_resource([f"do:droplet:{vm_id}"])
152
153 def validate_region(self, region):
154 if region is not None and region.upper() not in VALID_REGIONS:
155 fatal_error(f"Error: region {region} is not one of {VALID_REGIONS}")
156
157 def validate_image(self, image):
158 if image is not None and image not in VALID_IMAGES:
159 info(f"Warning: image {image} is not one of these known valid images: {VALID_IMAGES}")
160
161 def _get_do_ssh_key(self, name):
162 keys = self._manager.get_all_sshkeys()
163 for key in keys:
164 if key.name == name:
165 return key
166 return None
167
168 @property
169 def provider_name(self) -> str:
170 return "DigitalOcean"
171 {
"file_path": "/home/david/projects/stirlingbridge/machine/machine/providers/vultr.py"
}1 import base64
2 import time
3
4 from vultr import Vultr, VultrException
5
6 from machine.log import fatal_error, info
7 from machine.provider import CloudProvider, VM, SSHKey, DNSRecord
8
9
10 VALID_REGIONS = [
11 "ewr", "ord", "dfw", "sea", "lax", "atl", "ams", "lhr", "fra",
12 "sjc", "syd", "nrt", "cdg", "icn", "mia", "sgp", "sto",
13 "mex", "mad", "sao", "del", "hnl", "yto", "blr", "jnb",
14 "bom", "tlv",
15 ]
16
17
18 def _instance_to_vm(instance) -> VM:
19 return VM(
20 id=instance.get("id", ""),
21 name=instance.get("label", ""),
22 tags=instance.get("tags", []),
23 region=instance.get("region", ""),
24 ip_address=instance.get("main_ip", ""),
25 status=instance.get("status", ""),
26 )
27
28
29 class VultrProvider(CloudProvider):
30 def __init__(self, provider_config):
31 info("WARNING: Vultr support is experimental and has not been fully verified. Use with caution.")
32 if "api-key" not in provider_config:
33 fatal_error("Required key 'api-key' not found in 'vultr' section of config file")
34 self._api_key = provider_config["api-key"]
35 self._client = Vultr(self._api_key)
36
37 def create_vm(self, name, region, image, size, ssh_key_names, tags, user_data) -> VM:
38 sshkey_ids = []
39 for ssh_key_name in ssh_key_names:
40 ssh_key = self._get_vultr_ssh_key(ssh_key_name)
41 if not ssh_key:
42 fatal_error(f"Error: SSH key '{ssh_key_name}' not found in Vultr")
43 sshkey_ids.append(ssh_key["id"])
44
45 kwargs = {
46 "os_id": int(image),
47 "label": name,
48 "hostname": name,
49 "sshkey_id": sshkey_ids,
50 "tags": tags,
51 "backups": "disabled",
52 }
53 if user_data:
54 kwargs["user_data"] = base64.b64encode(user_data.encode()).decode()
55
56 try:
57 result = self._client.create_instance(region, size, **kwargs)
58 except VultrException as e:
59 fatal_error(f"Error creating instance: {e}")
60
61 return _instance_to_vm(result)
62
63 def get_vm(self, vm_id) -> VM:
64 try:
65 result = self._client.get_instance(vm_id)
66 except VultrException as e:
67 fatal_error(f"Error: machine with id {vm_id} not found: {e}")
68 return _instance_to_vm(result)
69
70 def destroy_vm(self, vm_id) -> bool:
71 # Vultr returns HTTP 500 if the instance is still pending or locked
72 # (e.g. during provisioning). Retry deletion with backoff.
73 for attempt in range(24):
74 try:
75 self._client.delete_instance(vm_id)
76 return True
77 except VultrException as e:
78 error_msg = str(e)
79 if "500" in error_msg and ("not currently active" in error_msg or "currently locked" in error_msg):
80 info("Waiting for instance to become ready before destroying...")
81 time.sleep(5)
82 elif "404" in error_msg:
83 return True # already gone
84 else:
85 fatal_error(f"Error: machine with id {vm_id} not found: {e}")
86 fatal_error(f"Error: timed out waiting to destroy instance {vm_id}")
87 return False
88
89 def list_vms(self, tag=None) -> list:
90 try:
91 params = {"tag": tag} if tag else None
92 result = self._client.list_instances(params=params)
93 except VultrException as e:
94 fatal_error(f"Error listing instances: {e}")
95 return [_instance_to_vm(i) for i in result]
96
97 def get_ssh_key(self, name) -> SSHKey:
98 key = self._get_vultr_ssh_key(name)
99 if not key:
100 return None
101 return SSHKey(
102 id=key["id"],
103 name=key["name"],
104 fingerprint=key.get("fingerprint", ""),
105 public_key=key.get("ssh_key", ""),
106 )
107
108 def list_ssh_keys(self) -> list:
109 try:
110 result = self._client.list_keys()
111 except VultrException as e:
112 fatal_error(f"Error listing SSH keys: {e}")
113 return [
114 SSHKey(
115 id=k["id"],
116 name=k["name"],
117 fingerprint=k.get("fingerprint", ""),
118 public_key=k.get("ssh_key", ""),
119 )
120 for k in result
121 ]
122
123 def create_dns_record(self, zone, record_type, name, data, ttl, tag=None) -> str:
124 try:
125 result = self._client.post(
126 f"/domains/{zone}/records",
127 type=record_type, name=name, data=data, ttl=ttl,
128 )
129 except VultrException:
130 info(f"Warning: DNS zone '{zone}' not found in Vultr, DNS record not set")
131 return None
132 record = result.get("record", result)
133 return record.get("id")
134
135 def delete_dns_record(self, zone, record_name) -> bool:
136 records = self.get_dns_records(zone)
137 for record in records:
138 if record.name == record_name:
139 try:
140 self._client.delete(f"/domains/{zone}/records/{record.id}")
141 except VultrException:
142 return False
143 return True
144 return False
145
146 def get_dns_records(self, zone) -> list:
147 try:
148 result = self._client.get(f"/domains/{zone}/records")
149 except VultrException:
150 info(f"Warning: DNS zone '{zone}' not found in Vultr")
151 return []
152 records = result.get("records", [])
153 return [
154 DNSRecord(
155 id=str(r.get("id", "")),
156 name=r.get("name", ""),
157 type=r.get("type", ""),
158 data=r.get("data", ""),
159 ttl=r.get("ttl", 0),
160 )
161 for r in records
162 ]
163
164 def list_domains(self) -> list:
165 try:
166 result = self._client.get("/domains")
167 except VultrException as e:
168 fatal_error(f"Error listing domains: {e}")
169 domains = result.get("domains", [])
170 return [d.get("domain", "") for d in domains]
171
172 def validate_region(self, region):
173 if region is not None and region.lower() not in VALID_REGIONS:
174 fatal_error(f"Error: region {region} is not one of {VALID_REGIONS}")
175
176 def validate_image(self, image):
177 try:
178 int(image)
179 except (ValueError, TypeError):
180 info(f"Warning: Vultr image (os_id) should be a numeric ID. Got: {image}")
181
182 def _get_vultr_ssh_key(self, name):
183 try:
184 result = self._client.list_keys()
185 except VultrException as e:
186 fatal_error(f"Error listing SSH keys: {e}")
187 for key in result:
188 if key.get("name") == name:
189 return key
190 return None
191
192 @property
193 def provider_name(self) -> str:
194 return "Vultr"
195 1 import sys 2 3 4 def fatal_error(s: str): 5 print(s, file=sys.stderr) 6 sys.exit(1) 7 8 9 def debug(s: str): 10 print(s, file=sys.stderr) 11 12 13 def info(s: str): 14 print(s, file=sys.stderr) 15 16 17 def output(s: str): 18 print(s) 19
grep -n "scripts\|entry\|main:main\|console" pyproject.toml; echo "---GIT LOG BRANCH---"; git log --oneline main..HEAD; echo "---DIFF---"; git diff main..HEAD --stat
1 # Exposing the truth that dependency injection is just a fancy name for global variables 2 3 from machine.types import CliOptions 4 5 6 class d: 7 opt: CliOptions = None 8
1 from abc import ABC, abstractmethod 2 from dataclasses import dataclass 3 4 5 @dataclass 6 class VM: 7 id: str 8 name: str 9 tags: list 10 region: str 11 ip_address: str 12 status: str 13 14 15 @dataclass 16 class SSHKey: 17 id: str 18 name: str 19 fingerprint: str 20 public_key: str 21 22 23 @dataclass 24 class DNSRecord: 25 id: str 26 name: str 27 type: str 28 data: str 29 ttl: int 30 31 32 class CloudProvider(ABC): 33 @abstractmethod 34 def create_vm(self, name, region, image, size, ssh_key_names, tags, user_data) -> VM: 35 pass 36 37 @abstractmethod 38 def get_vm(self, vm_id) -> VM: 39 pass 40 41 @abstractmethod 42 def destroy_vm(self, vm_id) -> bool: 43 pass 44 45 @abstractmethod 46 def list_vms(self, tag=None) -> list: 47 pass 48 49 @abstractmethod 50 def get_ssh_key(self, name) -> SSHKey: 51 pass 52 53 @abstractmethod 54 def list_ssh_keys(self) -> list: 55 pass 56 57 @abstractmethod 58 def create_dns_record(self, zone, record_type, name, data, ttl, tag=None) -> str: 59 pass 60 61 @abstractmethod 62 def delete_dns_record(self, zone, record_name) -> bool: 63 pass 64 65 @abstractmethod 66 def get_dns_records(self, zone) -> list: 67 pass 68 69 @abstractmethod 70 def list_domains(self) -> list: 71 pass 72 73 def list_projects(self) -> list: 74 return [] 75 76 def assign_to_project(self, project_name, vm_id): 77 pass 78 79 def validate_region(self, region): 80 pass 81 82 def validate_image(self, image): 83 pass 84 85 @property 86 def provider_name(self) -> str: 87 return self.__class__.__name__ 88
1 from dataclasses import dataclass 2 3 TAG_MACHINE_CREATED = "machine:created" 4 TAG_MACHINE_TYPE_PREFIX = "machine:type:" 5 TAG_MACHINE_SESSION_PREFIX = "machine:session:" 6 7 8 @dataclass 9 class CliOptions: 10 debug: bool 11 quiet: bool 12 verbose: bool 13 dry_run: bool 14 15 16 @dataclass 17 class Config: 18 provider_name: str 19 provider_config: dict 20 ssh_keys: list 21 dns_zone: str 22 machine_size: str 23 image: str 24 region: str 25 project: str 26 27 28 @dataclass 29 class MainCmdCtx: 30 config: Config 31 session_id: str 32 provider: object # CloudProvider instance 33 34 35 @dataclass 36 class MachineConfig: 37 new_user_name: str 38 script_url: str 39 script_dir: str 40 script_path: str 41 script_args: str 42
uv run python -c "
import digitalocean, inspect
import digitalocean.baseapi as b
print('DataReadError module:', digitalocean.DataReadError.__module__)
print('MRO:', [c.__name__ for c in digitalocean.DataReadError.__mro__])
print('has Error attr on pkg:', hasattr(digitalocean, 'Error'))
print('baseapi exceptions:', [n for n,o in vars(b).items() if isinstance(o,type) and issubclass(o,Exception)])
from vultr import VultrError, VultrException
print('Vultr MRO:', [c.__name__ for c in VultrException.__mro__])
" 2>&1 | head -40Traceback (most recent call last): File "<string>", line 8, in <module> ImportError: cannot import name 'VultrError' from 'vultr' (/home/david/projects/stirlingbridge/machine/.venv/lib/python3.12/site-packages/vultr/__init__.py) DataReadError module: digitalocean MRO: ['DataReadError', 'Error', 'Exception', 'BaseException', 'object'] has Error attr on pkg: True baseapi exceptions: ['Error', 'TokenError', 'DataReadError', 'JSONReadError', 'NotFoundError', 'EndPointError']
uv run python -c "
from vultr import Vultr, VultrException
print('VultrException MRO:', [c.__name__ for c in VultrException.__mro__])
import vultr, vultr.v1_server as vs
print('vultr exports:', [n for n in dir(vultr) if 'Exc' in n or 'Err' in n])
"Exit code 1 Traceback (most recent call last): File "<string>", line 4, in <module> ModuleNotFoundError: No module named 'vultr.v1_server' VultrException MRO: ['VultrException', 'Exception', 'BaseException', 'object']
{
"file_path": "/home/david/projects/stirlingbridge/machine/machine/providers/__init__.py"
}1 from machine.log import fatal_error
2 from machine.provider import CloudProvider
3
4 KNOWN_PROVIDERS = ["digital-ocean", "vultr", "gcp"]
5
6
7 def create_provider(provider_name, provider_config) -> CloudProvider:
8 if provider_name == "digital-ocean":
9 from machine.providers.digitalocean import DigitalOceanProvider
10
11 return DigitalOceanProvider(provider_config)
12 elif provider_name == "vultr":
13 from machine.providers.vultr import VultrProvider
14
15 return VultrProvider(provider_config)
16 elif provider_name == "gcp":
17 from machine.providers.gcp import GcpProvider
18
19 return GcpProvider(provider_config)
20 else:
21 fatal_error(f"Unknown provider: '{provider_name}'. Known providers: {', '.join(KNOWN_PROVIDERS)}")
22 ls /home/david/projects/stirlingbridge/machine/tests 2>/dev/null && echo "---" && sed -n '1,90p' /home/david/projects/stirlingbridge/machine/machine/subcommands/list.py
E2E.md
__init__.py
__pycache__
test_cli_integration.py
test_cloud_config.py
test_config.py
test_e2e.py
test_input_validation.py
---
import click
import json
from machine.log import fatal_error, output
from machine.types import MainCmdCtx, TAG_MACHINE_TYPE_PREFIX, TAG_MACHINE_SESSION_PREFIX
from machine.util import get_machine_type, is_machine_created, is_same_session, vm_to_json_obj
def print_normal(vms):
for vm in vms:
region = vm.region if vm.region else "unknown"
output(f"{vm.name} ({vm.id}, {region}, {get_machine_type(vm)}): {vm.ip_address}")
def print_quiet(vms):
for vm in vms:
output(vm.id)
def print_json(vms):
output(json.dumps([vm_to_json_obj(v) for v in vms]))
def get_vms(command_context, id=None, name=None, tag=None, type=None, region=None, all=False):
provider = command_context.provider
vms = []
if id:
vm = provider.get_vm(id)
if vm:
vms.append(vm)
if all:
vms = provider.list_vms()
else:
vms = provider.list_vms(tag=TAG_MACHINE_SESSION_PREFIX + command_context.session_id)
# we can't combine most filters over the API, so we also filter ourselves
if name:
vms = filter(lambda v: v.name == name, vms)
if tag:
vms = filter(lambda v: tag in v.tags, vms)
if type:
vms = filter(lambda v: TAG_MACHINE_TYPE_PREFIX + type.lower() in v.tags, vms)
if region:
vms = filter(lambda v: v.region and region == v.region, vms)
if not all:
vms = filter(lambda v: is_machine_created(v) and is_same_session(command_context, v), vms)
return list(vms)
@click.command(help="List machines")
@click.option("--id", metavar="<MACHINE-ID>", help="Filter by id")
@click.option("--name", "-n", metavar="<MACHINE-NAME>", help="Filter by name")
@click.option("--tag", "-t", metavar="<TAG-TEXT>", help="Filter by tag")
@click.option("--type", "-m", metavar="<MACHINE-TYPE>", help="Filter by type")
@click.option("--region", "-r", metavar="<REGION>", help="Filter by region")
@click.option("--output", "-o", metavar="<FORMAT>", help="Output format")
@click.option(
"--all",
is_flag=True,
default=False,
help="All machines, including those not created by this tool or by other sessions",
)
@click.option("--quiet", "-q", is_flag=True, default=False, help="Only display machine IDs")
@click.option(
"--unique",
is_flag=True,
default=False,
help="Return an error if there is more than one match",
)
@click.pass_context
def command(context, id, name, tag, type, region, all, output, quiet, unique):
command_context: MainCmdCtx = context.obj
vms = get_vms(command_context, id, name, tag, type, region, all)
if unique and len(vms) > 1:
fatal_error(f"ERROR: --unique match required but {len(vms)} matches found.")
if output == "json":
print_json(vms)
elif quiet:
print_quiet(vms)
else:
print_normal(vms)sed -n '1,60p' /home/david/projects/stirlingbridge/machine/machine/providers/gcp.py 2>/dev/null; echo "=== test_cli_integration head ==="; sed -n '1,60p' /home/david/projects/stirlingbridge/machine/tests/test_cli_integration.py
import base64
import hashlib
import os
from machine.log import fatal_error, info
from machine.provider import CloudProvider, VM, SSHKey, DNSRecord
# Tags are stored as newline-separated values in instance metadata so that
# colon-bearing tag strings (e.g. "machine:created") round-trip unchanged.
_TAGS_METADATA_KEY = "machine-tags"
_USER_DATA_METADATA_KEY = "user-data"
_SSH_KEYS_METADATA_KEY = "ssh-keys"
_DEFAULT_OPERATION_TIMEOUT = 300
_GCP_SCOPES = ["https://www.googleapis.com/auth/cloud-platform"]
def _load_credentials(provider_config):
creds_file = provider_config.get("credentials-file")
if creds_file:
from google.oauth2 import service_account
path = os.path.expanduser(creds_file)
return service_account.Credentials.from_service_account_file(path, scopes=_GCP_SCOPES)
import google.auth
creds, _ = google.auth.default(scopes=_GCP_SCOPES)
return creds
def _fingerprint(public_key):
parts = public_key.strip().split()
if len(parts) < 2:
return ""
try:
data = base64.b64decode(parts[1])
except (ValueError, base64.binascii.Error):
return ""
digest = hashlib.md5(data).hexdigest()
return ":".join(digest[i : i + 2] for i in range(0, len(digest), 2))
def _parse_ssh_keys(raw):
keys = []
for line in (raw or "").splitlines():
line = line.strip()
if not line or ":" not in line:
continue
user, key_data = line.split(":", 1)
keys.append((user.strip(), key_data.strip()))
return keys
class GcpProvider(CloudProvider):
def __init__(self, provider_config):
if "project-id" not in provider_config:
fatal_error("Required key 'project-id' not found in 'gcp' section of config file")
self._project = provider_config["project-id"]
=== test_cli_integration head ===
"""Integration tests that run the machine CLI as a subprocess, the same way a user would."""
import os
import subprocess
import textwrap
import pytest
def run_machine(*args, env_override=None):
"""Run the machine CLI as a subprocess and return the result."""
env = os.environ.copy()
if env_override:
env.update(env_override)
result = subprocess.run(
["uv", "run", "machine"] + list(args),
capture_output=True,
text=True,
env=env,
)
return result
def write_config(path, content):
with open(path, "w") as f:
f.write(textwrap.dedent(content))
class TestVersionCommand:
"""Smoke test: the CLI runs and the version command works without any config."""
def test_version_runs(self):
result = run_machine("version")
assert result.returncode == 0
assert result.stdout.strip() # prints something
def test_help_runs(self):
result = run_machine("--help")
assert result.returncode == 0
assert "Usage" in result.stdout
class TestInfoCommand:
"""Tests for the info diagnostic subcommand."""
@pytest.fixture()
def config_dir(self, tmp_path):
return tmp_path
def test_info_shows_config_file_path(self, config_dir):
config_file = config_dir / "config.yml"
write_config(
config_file,
"""\
digital-ocean:
access-token: fake-token
ssh-key: test-key
machine-size: s-1vcpu-1gb
image: ubuntu-22-04-x64
region: nyc1
project: test-projectsed -n '60,220p' /home/david/projects/stirlingbridge/machine/tests/test_cli_integration.py
project: test-project
machines:
test-machine:
new-user-name: testuser
""",
)
result = run_machine("--config-file", str(config_file), "info")
assert result.returncode == 0
assert f"Config file: {config_file}" in result.stdout
def test_info_shows_config_file_contents(self, config_dir):
config_file = config_dir / "config.yml"
write_config(
config_file,
"""\
digital-ocean:
access-token: fake-token
ssh-key: test-key
machine-size: s-1vcpu-1gb
image: ubuntu-22-04-x64
region: nyc1
project: test-project
machines:
test-machine:
new-user-name: testuser
""",
)
result = run_machine("--config-file", str(config_file), "info")
assert result.returncode == 0
assert "access-token: fake-token" in result.stdout
assert "ssh-key: test-key" in result.stdout
def test_info_shows_session_id(self, config_dir):
config_file = config_dir / "config.yml"
write_config(
config_file,
"""\
digital-ocean:
access-token: fake-token
ssh-key: test-key
machine-size: s-1vcpu-1gb
image: ubuntu-22-04-x64
region: nyc1
project: test-project
machines:
test-machine:
new-user-name: testuser
""",
)
result = run_machine("--config-file", str(config_file), "info")
assert result.returncode == 0
assert "Session ID file:" in result.stdout
assert "Session ID:" in result.stdout
def test_info_shows_providers(self, config_dir):
config_file = config_dir / "config.yml"
write_config(
config_file,
"""\
digital-ocean:
access-token: fake-token
ssh-key: test-key
machine-size: s-1vcpu-1gb
image: ubuntu-22-04-x64
region: nyc1
project: test-project
machines:
test-machine:
new-user-name: testuser
""",
)
result = run_machine("--config-file", str(config_file), "info")
assert result.returncode == 0
assert "digital-ocean" in result.stdout
assert "vultr" in result.stdout
assert "Active provider: digital-ocean" in result.stdout
def test_info_shows_vultr_as_active_provider(self, config_dir):
config_file = config_dir / "config.yml"
write_config(
config_file,
"""\
vultr:
api-key: fake-key
ssh-key: test-key
machine-size: vc2-1c-1gb
image: ubuntu-22-04-x64
region: ewr
machines:
test-machine:
new-user-name: testuser
""",
)
result = run_machine("--config-file", str(config_file), "info")
assert result.returncode == 0
assert "Active provider: vultr" in result.stdout
class TestEnvVarExpansionIntegration:
"""End-to-end tests that verify environment variable expansion works
when the actual machine tool is invoked with a config file."""
@pytest.fixture()
def config_dir(self, tmp_path):
return tmp_path
def test_env_var_expanded_in_config(self, config_dir):
config_file = config_dir / "config.yml"
write_config(
config_file,
"""\
digital-ocean:
access-token: "${TEST_DO_TOKEN}"
ssh-key: test-key
machine-size: s-1vcpu-1gb
image: ubuntu-22-04-x64
region: nyc1
project: test-project
machines:
test-machine:
new-user-name: testuser
""",
)
result = run_machine(
"--debug",
"--config-file",
str(config_file),
"types",
env_override={"TEST_DO_TOKEN": "tok_secret_12345"},
)
assert result.returncode == 0
# The debug output on stderr should contain the expanded token value
assert "tok_secret_12345" in result.stderr
# And should NOT contain the unexpanded variable reference
assert "${TEST_DO_TOKEN}" not in result.stderr
# The types command should list the machine name on stdout
assert "test-machine" in result.stdout
def test_multiple_env_vars_expanded(self, config_dir):
config_file = config_dir / "config.yml"
write_config(
config_file,
"""\
digital-ocean:
access-token: "${TEST_TOKEN}"
ssh-key: "${TEST_SSH_KEY}"
machine-size: s-1vcpu-1gb
image: ubuntu-22-04-x64
region: nyc1
project: test-project
machines:
dev:
new-user-name: devuser
""",
)
result = run_machine(
"--debug",
"--config-file",
str(config_file),
"types",
env_override={"TEST_TOKEN": "expanded_token", "TEST_SSH_KEY": "expanded_key"},sed -n '1,40p' /home/david/projects/stirlingbridge/machine/tests/test_e2e.py; echo "=== how tests run in CI ==="; grep -rn "pytest\|test" /home/david/projects/stirlingbridge/machine/Makefile /home/david/projects/stirlingbridge/machine/sh/*.sh 2>/dev/null | head
"""End-to-end tests that create real VMs on a cloud provider.
These tests require a real cloud provider environment and are NOT run as part of
the normal test suite. They must be invoked explicitly:
uv run pytest tests/test_e2e.py -v
Provider selection:
E2E_PROVIDER - Provider name: "digital-ocean" (default), "vultr", or "gcp"
Required environment variables (all providers):
E2E_SSH_KEY - Name of an SSH key already registered with the provider
Required environment variables (DigitalOcean):
E2E_DO_TOKEN - DigitalOcean API token
E2E_DO_DNS_ZONE - DNS zone hosted at DigitalOcean (e.g. "do.example.com")
E2E_PROJECT - DO project name to assign droplets to
Required environment variables (Vultr):
E2E_VULTR_API_KEY - Vultr API key
E2E_VULTR_DNS_ZONE - DNS zone hosted at Vultr (e.g. "example.com")
Required environment variables (GCP):
E2E_GCP_PROJECT_ID - GCP project ID where test resources are created
E2E_GCP_DNS_ZONE - DNS zone hosted in Cloud DNS (e.g. "gcp.example.com")
Optional (GCP):
E2E_GCP_CREDENTIALS_FILE - Path to a service account JSON key file
(if unset, Application Default Credentials are used)
Optional environment variables:
E2E_REGION - Region slug (default: provider-specific)
E2E_IMAGE - Image slug or ID (default: provider-specific)
E2E_SIZE - Machine size slug (default: provider-specific)
"""
import json
import os
import subprocess
import uuid
import pytest
=== how tests run in CI ===
/home/david/projects/stirlingbridge/machine/Makefile:8:test:
/home/david/projects/stirlingbridge/machine/Makefile:9: uv run pytest tests/ -v
/home/david/projects/stirlingbridge/machine/Makefile:11:test-e2e:
/home/david/projects/stirlingbridge/machine/Makefile:12: uv run pytest tests/test_e2e.py -v -m e2esed -n '1,60p' /home/david/projects/stirlingbridge/machine/pyproject.toml
[project]
name = "machine"
version = "1.0.0"
description = "Utility for creating and managing VMs"
readme = "README.md"
license = "AGPL-3.0-only"
requires-python = ">=3.10"
authors = [
{ name = "Stirlingbridge", email = "info@stirlingbridge.website" },
]
classifiers = [
"Programming Language :: Python :: 3.10",
"Operating System :: OS Independent",
]
dependencies = [
"click==8.1.7",
"python-digitalocean==1.17.0",
"ruamel.yaml>=0.17.32",
"requests>=2.32.4",
"expandvars==1.0.0",
"vultr-python>=0.1.5",
"google-cloud-compute>=1.19.0",
"google-cloud-dns>=0.35.0",
]
[project.scripts]
machine = "machine.main:main"
[project.urls]
Homepage = "https://github.com/stirlingbridge/machine"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.pytest.ini_options]
markers = [
"e2e: end-to-end tests that create real resources on DigitalOcean (require E2E_* env vars)",
]
addopts = "-m 'not e2e'"
[dependency-groups]
dev = [
"ruff",
"pytest",
]
[tool.ruff]
line-length = 132
[tool.ruff.lint]
select = ["E", "F", "W", "C90"]
ignore = ["E203"]
[tool.ruff.lint.mccabe]
max-complexity = 25
[tool.ruff.lint.per-file-ignores]
"machine/cloud_config.py" = ["E501"]cd /home/david/projects/stirlingbridge/machine
cat > /tmp/claude-1000/-home-david-projects-stirlingbridge-machine/c97f0339-2e25-4b0f-b459-d70bc5b1e1fd/scratchpad/badcfg.yml <<'EOF'
digital-ocean:
access-token: definitely-not-a-valid-token
ssh-key: test-key
machine-size: s-1vcpu-1gb
image: ubuntu-22-04-x64
region: nyc1
project: test-project
machines:
test-machine:
new-user-name: testuser
EOF
uv run machine --config-file /tmp/claude-1000/-home-david-projects-stirlingbridge-machine/c97f0339-2e25-4b0f-b459-d70bc5b1e1fd/scratchpad/badcfg.yml list 2>&1 | tail -20; echo "EXIT: ${PIPESTATUS[0]}" return __callback(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/projects/stirlingbridge/machine/.venv/lib/python3.12/site-packages/click/decorators.py", line 33, in new_func
return f(get_current_context(), *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/projects/stirlingbridge/machine/machine/subcommands/list.py", line 81, in command
vms = get_vms(command_context, id, name, tag, type, region, all)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/projects/stirlingbridge/machine/machine/subcommands/list.py", line 36, in get_vms
vms = provider.list_vms(tag=TAG_MACHINE_SESSION_PREFIX + command_context.session_id)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/projects/stirlingbridge/machine/machine/providers/digitalocean.py", line 85, in list_vms
droplets = self._manager.get_all_droplets(tag_name=tag)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/projects/stirlingbridge/machine/.venv/lib/python3.12/site-packages/digitalocean/Manager.py", line 66, in get_all_droplets
data = self.get_data("droplets/", params=params)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/projects/stirlingbridge/machine/.venv/lib/python3.12/site-packages/digitalocean/baseapi.py", line 233, in get_data
raise DataReadError(msg)
digitalocean.DataReadError: Unable to authenticate you
EXIT: 1Reproduced. Network works and DO returns the auth error as DataReadError. Now I'll implement a top-level entry-point wrapper that converts provider API errors into clear messages.
main.add_command(check.command, "check") main.add_command(create.command, "create") main.add_command(destroy.command, "destroy") main.add_command(domains.command, "domains") main.add_command(info.command, "info") main.add_command(list.command, "list") main.add_command(list_domain.command, "list-domain") main.add_command(projects.command, "projects") main.add_command(ssh_keys.command, "ssh-keys") main.add_command(types.command, "types") main.add_command(status.command, "status")
main.add_command(check.command, "check")
main.add_command(create.command, "create")
main.add_command(destroy.command, "destroy")
main.add_command(domains.command, "domains")
main.add_command(info.command, "info")
main.add_command(list.command, "list")
main.add_command(list_domain.command, "list-domain")
main.add_command(projects.command, "projects")
main.add_command(ssh_keys.command, "ssh-keys")
main.add_command(types.command, "types")
main.add_command(status.command, "status")
def _provider_api_exception_types():
"""Base exception classes raised by the cloud provider SDKs on API failures.
Collected lazily so importing this module (and the common DigitalOcean path)
does not pull in every provider SDK up front.
"""
import digitalocean
from vultr import VultrException
types = [digitalocean.Error, VultrException]
try:
from google.api_core import exceptions as google_api_exceptions
from google.auth import exceptions as google_auth_exceptions
types.append(google_api_exceptions.GoogleAPICallError)
types.append(google_auth_exceptions.GoogleAuthError)
except ImportError:
pass
return tuple(types)
def _friendly_provider_error(e) -> str:
"""Turn a raw provider SDK exception into a clear, actionable message."""
detail = str(e).strip()
lowered = detail.lower()
auth_markers = (
"unable to authenticate",
"authentication",
"unauthenticated",
"unauthorized",
"invalid api key",
"permission",
"forbidden",
"401",
"403",
)
if any(marker in lowered for marker in auth_markers):
return (
"Error: the cloud provider rejected the request as unauthenticated or unauthorized.\n"
"Check that the API token/key in your config file is correct and has not expired."
)
return f"Error: cloud provider request failed: {detail}"
def cli():
"""Console-script entry point.
Wraps the Click group so that errors raised by a provider's API (for
example an expired or invalid access token) are reported as a clear
message instead of an uncaught Python traceback (#95). Pass --debug to
see the underlying traceback.
"""
try:
main()
except _provider_api_exception_types() as e:
if d.opt is not None and d.opt.debug:
raise
fatal_error(_friendly_provider_error(e))The file /home/david/projects/stirlingbridge/machine/machine/main.py has been updated successfully. (file state is current in your context — no need to Read it back)
from machine.log import output
from machine.log import fatal_error, output
machine = "machine.main:main"
machine = "machine.main:cli"
The file /home/david/projects/stirlingbridge/machine/machine/main.py has been updated successfully. (file state is current in your context — no need to Read it back)
The file /home/david/projects/stirlingbridge/machine/pyproject.toml has been updated successfully. (file state is current in your context — no need to Read it back)
Since the entry point changed, I need to reinstall so the console script points at cli. Let me sync and test.
cd /home/david/projects/stirlingbridge/machine
uv sync --quiet 2>&1 | tail -3
SCR=/tmp/claude-1000/-home-david-projects-stirlingbridge-machine/c97f0339-2e25-4b0f-b459-d70bc5b1e1fd/scratchpad/badcfg.yml
echo "=== list (invalid token) ==="; uv run machine --config-file $SCR list 2>&1; echo "EXIT: $?"
echo "=== list --debug (invalid token) ==="; uv run machine --config-file $SCR --debug list 2>&1 | tail -4; echo "EXIT: ${PIPESTATUS[0]}"=== list (invalid token) ===
Error: the cloud provider rejected the request as unauthenticated or unauthorized.
Check that the API token/key in your config file is correct and has not expired.
EXIT: 1
=== list --debug (invalid token) ===
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/david/projects/stirlingbridge/machine/.venv/lib/python3.12/site-packages/digitalocean/baseapi.py", line 233, in get_data
raise DataReadError(msg)
digitalocean.DataReadError: Unable to authenticate you
EXIT: 1Both paths work: friendly message by default, full traceback with --debug. Let me verify normal commands still work and Click's own errors are unaffected.
cd /home/david/projects/stirlingbridge/machine
SCR=/tmp/claude-1000/-home-david-projects-stirlingbridge-machine/c97f0339-2e25-4b0f-b459-d70bc5b1e1fd/scratchpad/badcfg.yml
echo "=== version ==="; uv run machine version; echo "EXIT: $?"
echo "=== types (no network) ==="; uv run machine --config-file $SCR types 2>&1 | head -3; echo "EXIT: ${PIPESTATUS[0]}"
echo "=== bad subcommand (Click usage error) ==="; uv run machine nope 2>&1 | head -3; echo "EXIT: ${PIPESTATUS[0]}"
echo "=== -h ==="; uv run machine -h >/dev/null 2>&1; echo "EXIT: $?"=== version === dev EXIT: 0 === types (no network) === test-machine EXIT: 0 === bad subcommand (Click usage error) === Usage: machine [OPTIONS] COMMAND [ARGS]... Try 'machine -h' for help. EXIT: 120 === -h === EXIT: 0
cd /home/david/projects/stirlingbridge/machine
cat > /tmp/claude-1000/-home-david-projects-stirlingbridge-machine/c97f0339-2e25-4b0f-b459-d70bc5b1e1fd/scratchpad/vultr.yml <<'EOF'
vultr:
api-key: not-a-valid-key
ssh-key: test-key
machine-size: vc2-1c-1gb
image: "1743"
region: ewr
machines:
test-machine:
new-user-name: testuser
EOF
uv run machine --config-file /tmp/claude-1000/-home-david-projects-stirlingbridge-machine/c97f0339-2e25-4b0f-b459-d70bc5b1e1fd/scratchpad/vultr.yml list 2>&1; echo "EXIT: ${PIPESTATUS[0]}"WARNING: Vultr support is experimental and has not been fully verified. Use with caution. Error listing instances: Error 502: We are currently conducting some software upgrades. Check back in a few minutes! EXIT: 1
Vultr's per-method handling already catches its exceptions (that 502 is a transient Vultr maintenance error, handled by the existing fatal_error). The DigitalOcean read paths were the real gap, and the top-level wrapper is a safety net for those and GCP.
Now let me add hermetic tests for the new behavior.
"""Tests for graceful handling of cloud provider API errors (#95).
These are hermetic: instead of making real API calls they drive the
``cli`` entry-point wrapper directly, forcing the underlying Click group to
raise the same exceptions a provider SDK would raise on an auth failure.
"""
import digitalocean
import pytest
from machine import main as main_module
from machine.types import CliOptions
@pytest.fixture(autouse=True)
def reset_options():
"""Ensure the global CLI options don't leak between tests."""
saved = main_module.d.opt
main_module.d.opt = None
yield
main_module.d.opt = saved
class TestFriendlyProviderError:
def test_auth_failure_message_is_actionable(self):
msg = main_module._friendly_provider_error(digitalocean.DataReadError("Unable to authenticate you"))
assert "unauthenticated or unauthorized" in msg
assert "config file" in msg
# The scary raw exception text should not leak into the friendly message.
assert "DataReadError" not in msg
def test_non_auth_failure_includes_detail(self):
msg = main_module._friendly_provider_error(digitalocean.DataReadError("Rate limit exceeded"))
assert "Rate limit exceeded" in msg
assert "cloud provider request failed" in msg
class TestCliWrapper:
def _run_cli(self, monkeypatch, exc):
def boom():
raise exc
monkeypatch.setattr(main_module, "main", boom)
return main_module.cli
def test_auth_error_produces_message_not_traceback(self, monkeypatch, capsys):
run = self._run_cli(monkeypatch, digitalocean.DataReadError("Unable to authenticate you"))
with pytest.raises(SystemExit) as exc_info:
run()
assert exc_info.value.code == 1
err = capsys.readouterr().err
assert "unauthenticated or unauthorized" in err
assert "Traceback" not in err
def test_debug_flag_reraises_original_exception(self, monkeypatch):
main_module.d.opt = CliOptions(debug=True, quiet=False, verbose=False, dry_run=False)
run = self._run_cli(monkeypatch, digitalocean.DataReadError("Unable to authenticate you"))
with pytest.raises(digitalocean.DataReadError):
run()
def test_non_provider_exception_is_not_swallowed(self, monkeypatch):
run = self._run_cli(monkeypatch, ValueError("some internal bug"))
with pytest.raises(ValueError):
run()
File created successfully at: /home/david/projects/stirlingbridge/machine/tests/test_error_handling.py (file state is current in your context — no need to Read it back)
cd /home/david/projects/stirlingbridge/machine uv run pytest tests/test_error_handling.py -q 2>&1 | tail -20 echo "=== lint ==="; uv run ruff check machine/ tests/test_error_handling.py 2>&1 | tail -20
All new tests pass and lint is clean. Let me run the full suite to confirm nothing else broke.
.................................................... [100%] 52 passed, 14 deselected in 6.62s
Let me check the shiv build script references the entry point correctly, since I changed it.
cd /home/david/projects/stirlingbridge/machine; grep -rn "main:main\|main:cli\|console_script\|entry\|-e \|--entry" sh/build-package.sh
cd /home/david/projects/stirlingbridge/machine; grep -rn "machine.main\|shiv\|-c \|-e " sh/build-package.sh
Done. Here's what I did for issue #95.
An invalid/expired provider API token caused an uncaught digitalocean.DataReadError to bubble up as a raw Python traceback. The DigitalOcean provider only wrapped a few calls (create_vm, destroy_vm), so read paths like list_vms → get_all_droplets crashed ungracefully.
Added a top-level entry-point wrapper cli() in machine/main.py that runs the Click group and catches the provider SDKs' base API exceptions, converting them into a clear message instead of a traceback:
digitalocean.Error, vultr.VultrException, and (lazily) the Google Cloud auth/API exception bases — so it's a safety net across all subcommands and providers, not just list.--debug re-raises the original exception so the full traceback is still available for troubleshooting.Updated [project.scripts] in pyproject.toml to point machine at machine.main:cli (the shiv build uses this console script automatically).
Now:
$ machine --config-file … list
Error: the cloud provider rejected the request as unauthenticated or unauthorized.
Check that the API token/key in your config file is correct and has not expired.
--debug still shows the traceback.version, types), Click usage errors, and help are unaffected.tests/test_error_handling.py (message formatting, wrapper catches provider errors, --debug re-raises, non-provider exceptions pass through).I haven't committed — let me know if you'd like me to.
Thanks! I've committed and pushed the changes on the current branch and made a PR at github. Could you please add your notes above to the PR?
cd /home/david/projects/stirlingbridge/machine; gh pr view --json number,title,url,body 2>&1 | head -40
cd /home/david/projects/stirlingbridge/machine; echo "=== open PRs ==="; gh pr list --state open --json number,title,headRefName,url 2>&1; echo "=== current branch ==="; git branch --show-current; echo "=== recent PRs any state ==="; gh pr list --state all --limit 5 --json number,title,headRefName,url 2>&1
=== open PRs ===
[]
=== current branch ===
dboreham/improve-auth-fail-behavior
=== recent PRs any state ===
[{"headRefName":"dboreham/multiple-ssh-keys","number":92,"title":"Add support for multiple ssh keys","url":"https://github.com/stirlingbridge/machine/pull/92"},{"headRefName":"dboreham/enable-gcp-ci","number":90,"title":"Enable GCP provider e2e tests in CI","url":"https://github.com/stirlingbridge/machine/pull/90"},{"headRefName":"dboreham/gcp-support","number":88,"title":"Add GCP provider support","url":"https://github.com/stirlingbridge/machine/pull/88"},{"headRefName":"dboreham/add-lint-ci-job","number":85,"title":"Add a linter CI job","url":"https://github.com/stirlingbridge/machine/pull/85"},{"headRefName":"dboreham/check-subcommand","number":83,"title":"Add check subcommand","url":"https://github.com/stirlingbridge/machine/pull/83"}]I don't see a PR for this branch yet. Let me double-check the remote and search more broadly.
cd /home/david/projects/stirlingbridge/machine echo "=== remotes ==="; git remote -v echo "=== branch pushed? ==="; git ls-remote --heads origin dboreham/improve-auth-fail-behavior 2>&1 echo "=== search PRs mentioning auth ==="; gh pr list --state all --search "auth in:title" --json number,title,headRefName,url 2>&1 echo "=== search issue 95 linked ==="; gh pr list --state all --limit 20 --json number,headRefName --jq '.[].headRefName' 2>&1
The user doesn't want to proceed with this tool use. The tool use was rejected (eg. if it was a file edit, the new_string was NOT written to the file). STOP what you are doing and wait for the user to tell you how to proceed. Note: The user's next message may contain a correction or preference. Pay close attention — if they explain what went wrong or how they'd prefer you to work, consider saving that to memory for future sessions.
Sorry, that wasn't quite right: I've pushed the changes on the current branch but I'd like you to make the PR with your notes.
cd /home/david/projects/stirlingbridge/machine gh pr create --base main --head dboreham/improve-auth-fail-behavior --title "Handle provider auth failures gracefully (#95)" --body "$(cat <<'EOF' ## Problem An invalid/expired provider API token caused an uncaught `digitalocean.DataReadError` to bubble up as a raw Python traceback (#95). The DigitalOcean provider only wrapped a few calls (`create_vm`, `destroy_vm`), so read paths like `list_vms` → `get_all_droplets` crashed ungracefully. ## Fix Added a top-level entry-point wrapper `cli()` in `machine/main.py` that runs the Click group and catches the provider SDKs' base API exceptions, converting them into a clear message instead of a traceback: - Catches `digitalocean.Error`, `vultr.VultrException`, and (lazily) the Google Cloud auth/API exception bases — so it's a safety net across **all** subcommands and providers, not just `list`. - Auth-style failures (401/403/"unable to authenticate"/permission) get an actionable message pointing at the config token; other API errors report the underlying detail. - `--debug` re-raises the original exception so the full traceback is still available for troubleshooting. - Non-provider exceptions (real bugs) are **not** swallowed — they still surface. Updated `[project.scripts]` in `pyproject.toml` to point `machine` at `machine.main:cli` (the shiv build uses this console script automatically). Now: ``` $ machine --config-file … list Error: the cloud provider rejected the request as unauthenticated or unauthorized. Check that the API token/key in your config file is correct and has not expired. ``` ## Verification - Reproduced the original traceback, then confirmed the graceful message (exit 1) after the fix, and that `--debug` still shows the traceback. - Confirmed normal commands (`version`, `types`), Click usage errors, and help are unaffected. - Added hermetic tests in `tests/test_error_handling.py` (message formatting, wrapper catches provider errors, `--debug` re-raises, non-provider exceptions pass through). - Full suite: **52 passed**, lint clean. 🤖 Generated with [Claude Code](https://claude.com/claude-code) EOF )" 2>&1
Created the PR: https://github.com/stirlingbridge/machine/pull/96
It targets main from dboreham/improve-auth-fail-behavior, titled "Handle provider auth failures gracefully (#95)", with the Problem / Fix / Verification notes in the body.
Ok thanks. Looking at the CI test results on the PR I noticed that the Digitalocean e2e test failed. The log shows that it failed with API auth errors so I initially suspected something related to the current code changes. However manual testing (using uv run on this machine) showed that a valid DO token worked. So then I checked the run history for that job (in addition to PR and main branch merge triggers, there is a scheduled weekly run) which showed that it had been failing for some time(!) somehow GitHub is not sending me the run failure notification emails. Anyway, this likely implies that the API token used by the CI environment has expired. I will issue a new one and re-run the failing job.
That diagnosis makes sense — the e2e job runs against real DO resources, so an expired CI token would fail with exactly those auth errors regardless of this PR. And notably, my change only wraps the CLI entry point; it doesn't touch the create/destroy API paths the e2e tests exercise, so it wouldn't cause (or fix) that failure.
Let me point you to exactly where the CI token is configured so the rotation is quick:
cd /home/david/projects/stirlingbridge/machine echo "=== workflow files ==="; ls .github/workflows/ echo "=== token / secret refs ==="; grep -rn -i "token\|secret\|E2E_DO\|schedule\|cron" .github/workflows/ 2>&1
=== workflow files ===
build-release.yml
e2e-test.yml
lint.yml
test.yml
=== token / secret refs ===
.github/workflows/e2e-test.yml:9: schedule:
.github/workflows/e2e-test.yml:11: - cron: "0 6 * * 1"
.github/workflows/e2e-test.yml:20: id-token: write
.github/workflows/e2e-test.yml:45: E2E_DO_TOKEN: ${{ secrets.E2E_DO_TOKEN }}
.github/workflows/e2e-test.yml:46: E2E_DO_DNS_ZONE: ${{ vars.E2E_DO_DNS_ZONE }}
.github/workflows/e2e-test.yml:49: E2E_VULTR_API_KEY: ${{ secrets.E2E_VULTR_API_KEY }}Here's what you'll need:
E2E_DO_TOKEN (referenced at .github/workflows/e2e-test.yml:45). The DNS zone is a repo variable (vars.E2E_DO_DNS_ZONE), so only the token needs updating.cron: "0 6 * * 1" — weekly, Mondays 06:00 UTC (line 11), which matches the recurring failures you found.Once you've issued the new DO token, you can rotate the secret and re-run without leaving the terminal:
# Set the new token (prompts for the value, doesn't echo it to history)
gh secret set E2E_DO_TOKEN --repo stirlingbridge/machine
# Re-run just the failed e2e job on this PR's latest run
gh run list --workflow e2e-test.yml --branch dboreham/improve-auth-fail-behavior --limit 1
gh run rerun <run-id> --failed
One heads-up on notifications: GitHub only emails scheduled-workflow failures to the user whose account last modified the workflow file, and it silently stops after ~repeated failures on a schedule. So the missing emails are expected behavior rather than a second bug — worth adding an explicit failure notification (Slack/email step) to the workflow if you want reliable alerts on the weekly run. Happy to draft that if useful.
Want me to do anything here, or are you set to handle the token rotation and re-run yourself?