Skip to content

Commit

Permalink
cli: add delete subcommand
Browse files Browse the repository at this point in the history
This subcommand can delete a build in GBP. It is disabled by default
(for now).
  • Loading branch information
enku committed Jul 19, 2024
1 parent 38ede36 commit e1143a7
Show file tree
Hide file tree
Showing 4 changed files with 173 additions and 0 deletions.
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ addmachine = "gentoo_build_publisher.cli.addmachine"
addrepo = "gentoo_build_publisher.cli.addrepo"
apikey = "gentoo_build_publisher.cli.apikey"
check = "gentoo_build_publisher.cli.check"
delete = "gentoo_build_publisher.cli.delete"
worker = "gentoo_build_publisher.cli.worker"

[project.entry-points."gentoo_build_publisher.graphql_schema"]
Expand Down
59 changes: 59 additions & 0 deletions src/gentoo_build_publisher/cli/delete.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
"""Delete a build"""

import argparse

from gbpcli import GBP
from gbpcli.subcommands import completers as comp
from gbpcli.types import Console

from gentoo_build_publisher import publisher
from gentoo_build_publisher.settings import Settings
from gentoo_build_publisher.types import Build

HELP = "Delete the given build" ""


def handler(args: argparse.Namespace, _gbp: GBP, console: Console) -> int:
"""Delete the given build"""
if not delete_enabled(Settings.from_environ()):
console.err.print("Cannot delete builds because this feature is disabled.")
return 1

build = Build(args.machine, args.number)
tags = ([""] if publisher.published(build) else []) + publisher.tags(build)

if tags and not args.force:
tag_type = "published" if "" in tags else "tagged"
console.err.print(f"Cannot delete a {tag_type} build.")
return 1

for tag in tags:
publisher.untag(build.machine, tag)
publisher.delete(build)

return 0


# pylint: disable=duplicate-code
def parse_args(parser: argparse.ArgumentParser) -> None:
"""Set subcommand arguments"""
parser.add_argument(
"-f",
"--force",
action="store_true",
default=False,
help="Force delete published and/or tagged build",
)
comp.set(
parser.add_argument("machine", metavar="MACHINE", help="name of the machine"),
comp.machines,
)
comp.set(
parser.add_argument("number", metavar="NUMBER", help="build number"),
comp.build_ids,
)


def delete_enabled(settings: Settings) -> bool:
"""Return True if manual deletes are enabled in settings"""
return settings.MANUAL_DELETE_ENABLE
1 change: 1 addition & 0 deletions src/gentoo_build_publisher/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ class Settings(BaseSettings):
WORKER_BACKEND: str = "celery"
API_KEY_ENABLE: bool = True
API_KEY_LENGTH: int = 24
MANUAL_DELETE_ENABLE: bool = False

# Celery worker backend config
WORKER_CELERY_CONCURRENCY: int = 1
Expand Down
112 changes: 112 additions & 0 deletions tests/test_cli_delete.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
"""Tests for the cli delete subcommand"""

# pylint: disable=missing-docstring
from argparse import ArgumentParser, Namespace

from gentoo_build_publisher import publisher
from gentoo_build_publisher.cli import delete
from gentoo_build_publisher.types import Build

from . import TestCase, setup
from .setup_types import Fixtures, SetupOptions


@setup.depends("builds")
def build_fixture(_options: SetupOptions, fixtures: Fixtures) -> Build:
builds: list[Build] = fixtures.builds
return builds[0]


@setup.depends(build_fixture)
def args_fixture(_options: SetupOptions, fixtures: Fixtures) -> Namespace:
return Namespace(machine="babette", number=fixtures.build.build_id, force=False)


@setup.depends(build_fixture)
def force_args_fixture(_options: SetupOptions, fixtures: Fixtures) -> Namespace:
return Namespace(machine="babette", number=fixtures.build.build_id, force=True)


@setup.requires(
"pulled_builds", "console", "gbp", build_fixture, args_fixture, force_args_fixture
)
class GBPChkTestCase(TestCase):
options = {
"builds": {"per_day": 6},
"environ": {"BUILD_PUBLISHER_MANUAL_DELETE_ENABLE": "true"},
}

def test_deletes_build(self) -> None:
fixtures = self.fixtures
build: Build = fixtures.build

self.assertTrue(publisher.pulled(build))
delete.handler(fixtures.args, fixtures.gbp, fixtures.console.console)

self.assertFalse(publisher.pulled(build))

def test_published_build(self) -> None:
fixtures = self.fixtures
build: Build = fixtures.build
publisher.publish(build)

status = delete.handler(fixtures.args, fixtures.gbp, fixtures.console.console)

self.assertEqual(status, 1)
stderr = self.fixtures.console.stderr.getvalue()
self.assertEqual(stderr, "Cannot delete a published build.\n")
self.assertTrue(publisher.pulled(build))

status = delete.handler(
fixtures.force_args, fixtures.gbp, fixtures.console.console
)
self.assertEqual(status, 0)
self.assertFalse(publisher.pulled(build))

def test_tagged_build(self) -> None:
fixtures = self.fixtures
build: Build = fixtures.build
publisher.tag(build, "testing")

status = delete.handler(fixtures.args, fixtures.gbp, fixtures.console.console)

self.assertEqual(status, 1)
stderr = self.fixtures.console.stderr.getvalue()
self.assertEqual(stderr, "Cannot delete a tagged build.\n")
self.assertTrue(publisher.pulled(build))

status = delete.handler(
fixtures.force_args, fixtures.gbp, fixtures.console.console
)
self.assertEqual(status, 0)
self.assertFalse(publisher.pulled(build))


@setup.requires("pulled_builds", "console", "gbp", build_fixture, args_fixture)
class DisabledDeletestTests(TestCase):
options = {"environ": {"BUILD_PUBLISHER_MANUAL_DELETE_ENABLE": "false"}}

def test_deletes_disabled(self) -> None:
fixtures = self.fixtures
build: Build = fixtures.build

status = delete.handler(fixtures.args, fixtures.gbp, fixtures.console.console)

self.assertEqual(status, 1)
stderr = self.fixtures.console.stderr.getvalue()
self.assertEqual(
stderr, "Cannot delete builds because this feature is disabled.\n"
)
self.assertTrue(publisher.pulled(build))


class CheckParseArgs(TestCase):
"""Tests for the parse_args callback"""

def test(self) -> None:
parser = ArgumentParser()
delete.parse_args(parser)

dests = [i.dest for i in parser._actions] # pylint: disable=protected-access

self.assertEqual(dests, ["help", "force", "machine", "number"])

0 comments on commit e1143a7

Please sign in to comment.