Browse Source

typer (#2)

Reviewed-on: https://git.sporestack.com/SporeStack/sporestack-python/pulls/2
Co-authored-by: Teran McKinney <sega01@go-beyond.org>
Co-committed-by: Teran McKinney <sega01@go-beyond.org>
master
Teran McKinney 3 weeks ago
parent
commit
4a0f0c5993
  1. 2
      .coveragerc
  2. 3
      Makefile
  3. 7
      mypy.ini
  4. 6
      setup.py
  5. 4
      src/sporestack/api_client.py
  6. 185
      src/sporestack/client.py
  7. 16
      src/sporestack/validate.py
  8. 9
      tests/client_test.py
  9. 8
      tests/validate_test.py

2
.coveragerc

@ -0,0 +1,2 @@
[report]
show_missing = True

3
Makefile

@ -6,8 +6,7 @@ test:
black --check .
isort --check .
flake8 .
mypy .
mypy --strict . || true
mypy --strict .
pytest --cov=sporestack --cov-fail-under=40 --cov-report=term --durations=3 --cache-clear
install:

7
mypy.ini

@ -1,4 +1,5 @@
[mypy]
exclude = build
[mypy-pytest.*]
ignore_missing_imports = True
@ -6,11 +7,5 @@ ignore_missing_imports = True
[mypy-sshpubkeys.*]
ignore_missing_imports = True
[mypy-aaargh.*]
ignore_missing_imports = True
[mypy-pyqrcode.*]
ignore_missing_imports = True
[mypy-setuptools.*]
ignore_missing_imports = True

6
setup.py

@ -42,10 +42,10 @@ setup(
package_dir={"": "src"},
package_data={"sporestack": ["py.typed"]},
install_requires=[
"pyqrcode",
"segno",
"requests[socks]>=2.22.0",
"aaargh",
"typer",
],
entry_points={"console_scripts": ["sporestack = sporestack.client:main"]},
entry_points={"console_scripts": ["sporestack = sporestack.client:cli"]},
zip_safe=False, # This is so py.typed gets included.
)

4
src/sporestack/api_client.py

@ -144,7 +144,6 @@ def launch(
affiliate_amount: Optional[int] = None,
affiliate_token: Optional[str] = None,
) -> Any:
validate.currency(currency)
validate.flavor(flavor)
validate.machine_id(machine_id)
validate.operating_system(operating_system)
@ -179,7 +178,6 @@ def topup(
affiliate_token: Optional[str] = None,
) -> Any:
validate.machine_id(machine_id)
validate.currency(currency)
validate.affiliate_amount(affiliate_amount)
json_params = {
@ -247,7 +245,6 @@ def settlement_token_enable(
) -> Any:
validate.settlement_token(settlement_token)
validate.cents(cents)
validate.currency(currency)
json_params = {
"settlement_token": settlement_token,
@ -267,7 +264,6 @@ def settlement_token_add(
) -> Any:
validate.settlement_token(settlement_token)
validate.cents(cents)
validate.currency(currency)
json_params = {
"settlement_token": settlement_token,

185
src/sporestack/client.py

@ -11,15 +11,15 @@ import time
from pathlib import Path
from typing import Any, Dict, Optional
import aaargh
import pyqrcode
import segno
import typer
from . import api_client
from .api_client import API_ENDPOINT
from .flavors import all_sporestack_flavors
from .version import __version__
cli = aaargh.App()
cli = typer.Typer()
logging.basicConfig(level=logging.INFO)
@ -44,38 +44,28 @@ one hour at the very most.
Resize your terminal and try again if QR code above is not readable.
Press ctrl+c to abort."""
message = premessage.format(uri)
qr = pyqrcode.create(uri)
print(qr.terminal(module_color="black", background="white", quiet_zone=1))
qr = segno.make(uri)
# This prints.
qr.terminal()
print(message)
print(f"Approximate price in USD: {usd}")
input("[Press enter once you have made payment.]")
@cli.cmd
@cli.cmd_arg("hostname")
@cli.cmd_arg("--days", type=int, required=True)
@cli.cmd_arg("--ssh-key", type=str, required=True)
@cli.cmd_arg("--operating-system", type=str, required=True)
@cli.cmd_arg("--flavor", type=str, default=DEFAULT_FLAVOR)
@cli.cmd_arg("--currency", type=str, default=None)
@cli.cmd_arg("--settlement-token", type=str, default=None)
@cli.cmd_arg("--region", type=str, default=None)
@cli.cmd_arg("--api-endpoint", type=str, default=API_ENDPOINT)
@cli.cmd_arg("--affiliate-amount", type=int, default=None)
@cli.cmd_arg("--affiliate-token", type=str, default=None)
@cli.command()
def launch(
hostname: str,
days: int,
ssh_key: str,
operating_system: str,
days: int = typer.Option(...),
ssh_key_file: Path = typer.Option(...),
operating_system: str = typer.Option(...),
flavor: str = DEFAULT_FLAVOR,
currency: Optional[str] = None,
settlement_token: Optional[str] = None,
region: Optional[str] = None,
api_endpoint: str = API_ENDPOINT,
affiliate_amount: Optional[str] = None,
affiliate_amount: Optional[int] = None,
affiliate_token: Optional[str] = None,
) -> str:
) -> None:
"""
Attempts to launch a server.
"""
@ -93,14 +83,13 @@ def launch(
message = f"{hostname} already created."
raise ValueError(message)
with open(ssh_key) as fp:
ssh_key = fp.read()
ssh_key = ssh_key_file.read_text()
machine_id = random_machine_id()
def create_vm():
create = api_client.launch
return create(
def create_vm() -> Any:
assert currency is not None
return api_client.launch(
machine_id=machine_id,
days=days,
flavor=flavor,
@ -163,26 +152,19 @@ def launch(
created_dict["api_endpoint"] = api_endpoint
save_machine_info(created_dict)
print(pretty_machine_info(created_dict), file=sys.stderr)
return json.dumps(created_dict, indent=4)
print(json.dumps(created_dict, indent=4))
@cli.cmd
@cli.cmd_arg("hostname")
@cli.cmd_arg("--days", type=int, required=True)
@cli.cmd_arg("--currency", type=str, default=None)
@cli.cmd_arg("--settlement-token", type=str, default=None)
@cli.cmd_arg("--api-endpoint", type=str, default=None)
@cli.cmd_arg("--affiliate-amount", type=int, default=None)
@cli.cmd_arg("--affiliate-token", type=str, default=None)
@cli.command()
def topup(
hostname: str,
days: int,
days: int = typer.Option(...),
currency: Optional[str] = None,
settlement_token: Optional[str] = None,
api_endpoint: Optional[str] = None,
affiliate_amount: Optional[str] = None,
affiliate_amount: Optional[int] = None,
affiliate_token: Optional[str] = None,
) -> int:
) -> None:
"""
tops up an existing vm.
"""
@ -203,8 +185,11 @@ def topup(
machine_id = machine_info["machine_id"]
if api_endpoint is None:
api_endpoint = machine_info["api_endpoint"]
assert api_endpoint is not None
def topup_vm():
def topup_vm() -> Any:
assert currency is not None
assert api_endpoint is not None
return api_client.topup(
machine_id=machine_id,
days=days,
@ -246,7 +231,7 @@ def topup(
machine_info["expiration"] = topped_dict["expiration"]
save_machine_info(machine_info, overwrite=True)
return machine_info["expiration"]
print(machine_info["expiration"])
def machine_info_path() -> Path:
@ -296,12 +281,13 @@ def get_machine_info(hostname: str) -> Dict[str, Any]:
if not json_file.exists():
raise ValueError(f"{hostname} does not exist in {directory} as {json_file}")
machine_info = json.loads(json_file.read_bytes())
assert isinstance(machine_info, dict)
if machine_info["vm_hostname"] != hostname:
raise ValueError("hostname does not match filename.")
return machine_info
def pretty_machine_info(info) -> str:
def pretty_machine_info(info: Dict[str, Any]) -> str:
msg = "Hostname: {}\n".format(info["vm_hostname"])
msg += "Machine ID (keep this secret!): {}\n".format(info["machine_id"])
if "ipv6" in info["network_interfaces"][0]:
@ -319,8 +305,8 @@ def pretty_machine_info(info) -> str:
return msg
@cli.cmd
def flavors():
@cli.command()
def flavors() -> None:
"""
List all available flavors.
"""
@ -328,8 +314,8 @@ def flavors():
print(all_sporestack_flavors[flavor])
@cli.cmd
def list():
@cli.command()
def list() -> None:
"""
List all locally known servers.
"""
@ -361,7 +347,6 @@ def list():
print(pretty_machine_info(info))
print()
return None
def remove(hostname: str) -> None:
@ -379,21 +364,17 @@ def machine_exists(hostname: str) -> bool:
return directory.joinpath(f"{hostname}.json").exists()
@cli.cmd(name="get-attribute")
@cli.cmd_arg("hostname")
@cli.cmd_arg("attribute")
def get_attribute(hostname, attribute):
@cli.command()
def get_attribute(hostname: str, attribute: str) -> None:
"""
Returns an attribute about the VM.
"""
machine_info = get_machine_info(hostname)
return machine_info[attribute]
print(machine_info[attribute])
@cli.cmd
@cli.cmd_arg("hostname")
@cli.cmd_arg("--api-endpoint", type=str, default=None)
def info(hostname, api_endpoint=None):
@cli.command()
def info(hostname: str, api_endpoint: Optional[str] = None) -> None:
"""
Info on the VM
"""
@ -401,15 +382,15 @@ def info(hostname, api_endpoint=None):
machine_id = machine_info["machine_id"]
if api_endpoint is None:
api_endpoint = machine_info["api_endpoint"]
return json.dumps(
api_client.info(machine_id=machine_id, api_endpoint=api_endpoint), indent=4
print(
json.dumps(
api_client.info(machine_id=machine_id, api_endpoint=api_endpoint), indent=4
)
)
@cli.cmd
@cli.cmd_arg("hostname")
@cli.cmd_arg("--api-endpoint", type=str, default=None)
def start(hostname, api_endpoint=None):
@cli.command()
def start(hostname: str, api_endpoint: Optional[str] = None) -> None:
"""
Boots the VM.
"""
@ -420,10 +401,8 @@ def start(hostname, api_endpoint=None):
return api_client.start(machine_id=machine_id, api_endpoint=api_endpoint)
@cli.cmd
@cli.cmd_arg("hostname")
@cli.cmd_arg("--api-endpoint", type=str, default=None)
def stop(hostname, api_endpoint=None):
@cli.command()
def stop(hostname: str, api_endpoint: Optional[str] = None) -> None:
"""
Immediately kills the VM.
"""
@ -434,10 +413,8 @@ def stop(hostname, api_endpoint=None):
return api_client.stop(machine_id=machine_id, api_endpoint=api_endpoint)
@cli.cmd
@cli.cmd_arg("hostname")
@cli.cmd_arg("--api-endpoint", type=str, default=None)
def delete(hostname, api_endpoint=None):
@cli.command()
def delete(hostname: str, api_endpoint: Optional[str] = None) -> None:
"""
Deletes the VM (most likely prematurely.
"""
@ -451,15 +428,11 @@ def delete(hostname, api_endpoint=None):
print(f"{hostname} was deleted.")
@cli.cmd(name="settlement-token-enable")
@cli.cmd_arg("token")
@cli.cmd_arg("--dollars", type=int, default=None, required=True)
@cli.cmd_arg("--currency", type=str, required=True)
@cli.cmd_arg("--api-endpoint", type=str, default=API_ENDPOINT)
@cli.command()
def settlement_token_enable(
token: str,
dollars: int,
currency: str,
dollars: int = typer.Option(...),
currency: str = typer.Option(...),
api_endpoint: str = API_ENDPOINT,
) -> None:
"""
@ -470,7 +443,7 @@ def settlement_token_enable(
cents = dollars * 100
def enable_token():
def enable_token() -> Any:
return api_client.settlement_token_enable(
settlement_token=token,
cents=cents,
@ -498,24 +471,20 @@ def settlement_token_enable(
print(f"{token} has been enabled. Save it and don't lose it!")
@cli.cmd(name="settlement-token-add")
@cli.cmd_arg("token")
@cli.cmd_arg("--dollars", type=int, required=True)
@cli.cmd_arg("--currency", type=str, required=True)
@cli.cmd_arg("--api-endpoint", type=str, default=API_ENDPOINT)
@cli.command()
def settlement_token_add(
token: str,
dollars: int,
currency: str,
dollars: int = typer.Option(...),
currency: str = typer.Option(...),
api_endpoint: str = API_ENDPOINT,
):
) -> None:
"""
Adds balance to an existing settlement token.
"""
cents = dollars * 100
def add_to_token():
def add_to_token() -> Any:
return api_client.settlement_token_add(
token,
cents,
@ -541,49 +510,35 @@ def settlement_token_add(
if add_dict["paid"] is True:
break
return True
@cli.cmd(name="settlement-token-balance")
@cli.cmd_arg("token")
@cli.cmd_arg("--api-endpoint", type=str, default=API_ENDPOINT)
def settlement_token_balance(token: str, api_endpoint: str = API_ENDPOINT):
@cli.command()
def settlement_token_balance(token: str, api_endpoint: str = API_ENDPOINT) -> None:
"""
Gets balance for a settlement token.
"""
return api_client.settlement_token_balance(
settlement_token=token, api_endpoint=api_endpoint
)["usd"]
print(
api_client.settlement_token_balance(
settlement_token=token, api_endpoint=api_endpoint
)["usd"]
)
@cli.cmd(name="settlement-token-generate")
def settlement_token_generate() -> str:
@cli.command()
def settlement_token_generate() -> None:
"""
Generates a settlement token that can be enabled.
"""
return random_machine_id()
print(random_machine_id())
@cli.cmd
def version() -> str:
@cli.command()
def version() -> None:
"""
Returns the installed version.
"""
return __version__
def main() -> None:
output = cli.run()
if output is True:
exit(0)
elif output is False:
exit(1)
elif output is None:
exit(0)
else:
print(output)
print(__version__)
if __name__ == "__main__":
main()
cli()

16
src/sporestack/validate.py

@ -122,22 +122,6 @@ def is_unsigned_int(variable: int) -> bool:
return True
def expiration(expiration: int) -> None:
"""
Makes sure expiration is valid.
"""
if not is_unsigned_int(expiration):
raise TypeError("expiration must be an unsigned integer.")
def currency(currency: str) -> None:
"""
Makes sure currency is valid.
"""
if not isinstance(currency, str):
raise TypeError("currency must be None or str.")
def region(region: Optional[str]) -> None:
if region is None:
return

9
tests/client_test.py

@ -1,9 +0,0 @@
from sporestack import client
def test_random_machine_id() -> None:
assert len(client.random_machine_id()) == 64
def test_version() -> None:
assert "." in client.version()

8
tests/validate_test.py

@ -187,6 +187,14 @@ def test_region() -> None:
validate.region(True) # type: ignore
with pytest.raises(ValueError):
validate.region("")
# Not too big.
validate.region("a" * 200)
# Too big.
with pytest.raises(ValueError):
validate.region("a" * 201)
# Bad character.
with pytest.raises(ValueError):
validate.region("🥔")
def test_affiliate_amount() -> None:

Loading…
Cancel
Save