Source code for pcvs.cli.cli_session

import os
from datetime import datetime, timedelta

from rich.table import Table

from pcvs import NAME_BUILDFILE, io
from pcvs.backend import session as pvSession
from pcvs.helpers import utils

try:
    import rich_click as click
    click.rich_click.SHOW_ARGUMENTS = True
except ImportError:
    import click

from click.shell_completion import CompletionItem


[docs]def compl_session_token(ctx, args, incomplete) -> list: """Session name completion function. :param ctx: Click context :type ctx: :class:`Click.Context` :param args: the option/argument requesting completion. :type args: str :param incomplete: the user input :type incomplete: str """ sessions = pvSession.list_alive_sessions() if sessions is None: return [] return [CompletionItem(k, help=str(pvSession.Session.State(v['state']))) for k, v in sessions.items() if incomplete in str(k)]
@click.command(name="session", short_help="Manage multiple validations") @click.option('-c', '--clear', 'ack', type=int, default=None, help="Clear a 'completed' remote session, for removing from logs") @click.option('-C', '--clear-all', 'ack_all', is_flag=True, default=False, help="Clear all completed sessions, for removing from logs") @click.option('-l', '--list', is_flag=True, help="List detached sessions") @click.pass_context def session(ctx, ack, list, ack_all): """Manage sessions by listing or acknowledging their completion.""" sessions = pvSession.list_alive_sessions() if sessions is None: sessions = dict() if ack_all is True: for session_id in sessions.keys(): if sessions[session_id]['state'] != pvSession.Session.State.IN_PROGRESS: pvSession.remove_session_from_file(session_id) lockfile = os.path.join( sessions[session_id]['path'], NAME_BUILDFILE) utils.unlock_file(lockfile) elif ack is not None: if ack not in sessions.keys(): raise click.BadOptionUsage( '--ack', "No such Session id (see pcvs session)") elif sessions[ack]['state'] not in [pvSession.Session.State.ERROR, pvSession.Session.State.COMPLETED]: raise click.BadOptionUsage( '--ack', "This session is not completed yet") pvSession.remove_session_from_file(ack) lockfile = os.path.join(sessions[ack]['path'], NAME_BUILDFILE) utils.unlock_file(lockfile) else: # listing is the defualt if len(sessions) <= 0: io.console.print("[italic bold]No sessions") return table = Table(title="Sessions", expand=True) table.add_column("SID", justify="center", max_width=10) table.add_column("Status", justify="right") table.add_column("Started", justify="center") table.add_column("Elasped", justify="right") table.add_column("Location", justify="left") for sk, sv in sessions.items(): s = pvSession.Session() s.load_from(sk, sv) status = "Broken" duration = timedelta() line_style = "default" if s.state == pvSession.Session.State.IN_PROGRESS: duration = datetime.now() - s.property('started') status = "{:3.2f} %".format(s.property('progress')) line_style = "yellow" elif s.state == pvSession.Session.State.COMPLETED: duration = s.property('ended') - s.property('started') status = "100.00 %" line_style = "green bold" elif s.state == pvSession.Session.State.WAITING: duration = datetime.now() - s.property('started') status = "Waiting" line_style = "yellow" table.add_row("[{}]{:0>6}".format(line_style, s.id), "[{}]{}".format(line_style, status), "[{}]{}".format(line_style, datetime.strftime( s.property("started"), "%Y-%m-%d %H:%M")), "[{}]{}".format("red bold" if duration.days > 0 else line_style, timedelta(days=duration.days, seconds=duration.seconds)), "[{}]{}".format(line_style, s.property("path")) ) io.console.print(table)