Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 28 additions & 3 deletions tanco/driver.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,25 +2,25 @@
"""
command-line driver for tanco client.
"""
import argparse
import asyncio
import cmd as cmdlib
import os
import shlex
import sqlite3
import subprocess
import sys
import webbrowser
import pprint

import jwt as jwtlib
import websockets as w

from . import database as db
from . import model as m
from . import orgtest, runner
from .client import TancoClient
from .model import Config, TestDescription

Check failure on line 23 in tanco/driver.py

View workflow job for this annotation

GitHub Actions / test

ruff (I001)

tanco/driver.py:5:1: I001 Import block is un-sorted or un-formatted help: Organize imports

class TancoDriver(cmdlib.Cmd):
prompt = 'tanco> '
Expand All @@ -46,7 +46,12 @@
return
sid = db.get_server_id(self.client.url)
pre = self.client.get_pre_token()
webbrowser.open(self.client.url + 'auth/login?pre=' + pre)
url = self.client.url + 'auth/login?pre=' + pre
# Always print the URL so headless / remote users can open it manually.
print('Open this URL to log in:')
print(url)
if not webbrowser.open(url):
print('(No browser available; open the URL above in any browser.)')
jwt = self.client.get_jwt(pre=pre)
data = jwtlib.JWT().decode(jwt, do_verify=False) # TODO: verify
uid = db.uid_from_tokendata(sid, data['authid'], data['username'])
Expand Down Expand Up @@ -154,18 +159,18 @@
num_desc_removed = sum(1 for line in lines if line.strip().startswith(': '))

# Show summary
has_todo = any(l.strip().lower().startswith('#+todo:') for l in lines)

Check failure on line 162 in tanco/driver.py

View workflow job for this annotation

GitHub Actions / test

ruff (E741)

tanco/driver.py:162:68: E741 Ambiguous variable name: `l`
print(f'{"Would migrate" if args.check else "Migrating"}: {args.file}')
print(f'Changes:')

Check failure on line 164 in tanco/driver.py

View workflow job for this annotation

GitHub Actions / test

ruff (F541)

tanco/driver.py:164:15: F541 f-string without any placeholders help: Remove extraneous `f` prefix
print(f' - Add #+tanco-format: 0.2 directive')

Check failure on line 165 in tanco/driver.py

View workflow job for this annotation

GitHub Actions / test

ruff (F541)

tanco/driver.py:165:15: F541 f-string without any placeholders help: Remove extraneous `f` prefix
if not has_todo:
print(f' - Add #+todo: TODO | DONE TEST directive')

Check failure on line 167 in tanco/driver.py

View workflow job for this annotation

GitHub Actions / test

ruff (F541)

tanco/driver.py:167:19: F541 f-string without any placeholders help: Remove extraneous `f` prefix
print(f' - Migrate {num_tests} tests')
print(f' - Remove {num_name_removed} #+name: directives')
print(f' - Remove TODO/DONE from test headlines')

Check failure on line 170 in tanco/driver.py

View workflow job for this annotation

GitHub Actions / test

ruff (F541)

tanco/driver.py:170:15: F541 f-string without any placeholders help: Remove extraneous `f` prefix
print(f' - Remove {num_title_removed} = title lines')
print(f' - Remove {num_desc_removed} : description lines')
print(f' - Move description text to test bodies')

Check failure on line 173 in tanco/driver.py

View workflow job for this annotation

GitHub Actions / test

ruff (F541)

tanco/driver.py:173:15: F541 f-string without any placeholders help: Remove extraneous `f` prefix
print()

if args.check:
Expand Down Expand Up @@ -229,7 +234,7 @@
try:
arg = self.result[int(arg)]['name']
except IndexError:
print(f"Sorry, challenge number {arg} not found.")

Check failure on line 237 in tanco/driver.py

View workflow job for this annotation

GitHub Actions / test

ruff (Q000)

tanco/driver.py:237:23: Q000 Double quotes found but single quotes preferred help: Replace double quotes with single quotes
return
if arg == '--local':
print('starting local challenge.')
Expand Down Expand Up @@ -267,7 +272,16 @@
def do_show(self, arg=None):
"""Show a description of the current test."""
cfg = runner.load_config()
state = db.current_state(cfg.attempt)
if not cfg.attempt:
print('No active attempt.')
print('Use `tanco init` to start a new attempt.')
return
try:
state = db.current_state(cfg.attempt)
except IndexError:
print('No active attempt.')
print('Use `tanco init` to start a new attempt.')
return
if state == 'start':
print('You have not started the challenge yet.')
print('Use `tanco check` to make sure your program runs.')
Expand Down Expand Up @@ -321,7 +335,16 @@
# TODO: double check that all tests pass and repo is clean

cfg = runner.load_config()
state = db.current_state(cfg.attempt)
if not cfg.attempt:
print('No active attempt.')
print('Use `tanco init` to start a new attempt.')
return
try:
state = db.current_state(cfg.attempt)
except IndexError:
print('No active attempt.')
print('Use `tanco init` to start a new attempt.')
return
if state == 'done':
print('You have already completed the challenge!')
return
Expand Down Expand Up @@ -359,13 +382,15 @@
self.do_show()

def do_spawn(self, _arg):
"""Spawn the target program as a persistent process"""
self.target = runner.spawn(runner.load_config())

def ensure_target(self):
if not self.target:
self.do_spawn('')

def do_send(self, msg, suppress=False):
"""Send a message to the target program and print its output"""
self.ensure_target()
if msg.endswith(self.end_cmd): msg = msg[:-1]
self.target.stdin.write(msg + f'\n{self.end_cmd}\n')
Expand Down
76 changes: 76 additions & 0 deletions tests/test_basic.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,3 +73,79 @@ def test_valid_test_names_accepted(self):
self.assertEqual(len(c.tests), 2)
self.assertEqual(c.tests[0].name, 'foo.bar-1')
self.assertEqual(c.tests[1].name, 'baz_2.A')



class DriverNextTest(unittest.TestCase):
"""Regression: tanco next/show must not IndexError when no attempt is active."""

def setUp(self) -> None:
TANCO_SDB_PATH.unlink(missing_ok=True)
tanco.database.ensure_sdb()
with tanco.database.begin() as conn:
cur = conn.execute(
'insert into servers (url, name, info) values (?, ?, ?)',
(TANCO_SERVER, 'fakeserver', 'fake'))
sid = cur.lastrowid
cur = conn.execute(
'insert into users (sid, authid, username) values (?, ?, ?)',
(sid, 'fakeauthid', 'fakeuser'))
uid = cur.lastrowid
conn.execute('insert into tokens (uid, jwt) values (?, ?)', (uid, 'fakejwt'))
self._old_cwd = os.getcwd()

def tearDown(self) -> None:
os.chdir(self._old_cwd)
TANCO_SDB_PATH.unlink(missing_ok=True)

def _run_in_empty_project(self, method_name: str) -> str:
import io
from contextlib import redirect_stdout

import tanco.driver

buf = io.StringIO()
with tempfile.TemporaryDirectory() as tmp:
os.chdir(tmp)
d = tanco.driver.TancoDriver()
with redirect_stdout(buf):
getattr(d, method_name)('')
os.chdir(self._old_cwd)
return buf.getvalue()

def test_next_no_active_attempt_prints_friendly_message(self) -> None:
out = self._run_in_empty_project('do_next')
self.assertIn('No active attempt', out)
self.assertIn('tanco init', out)

def test_show_no_active_attempt_prints_friendly_message(self) -> None:
out = self._run_in_empty_project('do_show')
self.assertIn('No active attempt', out)
self.assertIn('tanco init', out)

def test_login_prints_oauth_url(self) -> None:
import io
from contextlib import redirect_stdout
from unittest.mock import MagicMock, patch

import tanco.driver

d = tanco.driver.TancoDriver()
d.client = MagicMock()
d.client.whoami.return_value = None
d.client.url = 'https://example.test/'
d.client.get_pre_token.return_value = 'PRETOKEN'
d.client.get_jwt.side_effect = RuntimeError('stop-after-print')

buf = io.StringIO()
with patch('tanco.driver.webbrowser.open', return_value=False) as open_mock, \
patch('tanco.driver.db.get_server_id', return_value=1):
with redirect_stdout(buf):
try:
d.do_login('')
except RuntimeError as e:
self.assertEqual(str(e), 'stop-after-print')

out = buf.getvalue()
self.assertIn('https://example.test/auth/login?pre=PRETOKEN', out)
open_mock.assert_called_once_with('https://example.test/auth/login?pre=PRETOKEN')
Loading