Skip to content
Merged
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
19 changes: 19 additions & 0 deletions s31/main.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import argparse
from collections import Counter
import sys
import os
import json
Expand Down Expand Up @@ -149,13 +150,17 @@ def config_argument(p):

try:
args.action(args)
except SystemExit:
raise # Re-raise SystemExit so it propagates properly
except RuntimeError as e:
print(e, file=sys.stderr)


def command_action(args):
try:
return do_command_action(args)
except SystemExit:
raise # Re-raise SystemExit so it propagates
finally:
if args.when_done_set is not None:
set_key(*args.when_done_set)
Expand Down Expand Up @@ -190,6 +195,8 @@ def do_command_action(args):
cmd_to_use = cmd.replace(assignment)
commands.append((screen_name, cmd_to_use, assignment))

_check_screen_name_non_overlap(commands)

if args.dry_run:
for screen_name, cmd_to_use, _ in commands:
if not args.sync:
Expand Down Expand Up @@ -243,6 +250,18 @@ def launch_worker(worker_idx, assignment, output_file, token):
notify(config, cmd_to_use, runner)


def _check_screen_name_non_overlap(commands):
command_counts = Counter(name for name, _, _ in commands)
duplicates = [name for name, count in command_counts.items() if count > 1]
if not duplicates:
return
print("Screen names generated multiple times:", file=sys.stderr)
for name in duplicates:
print(f" {name!r}", file=sys.stderr)
print("ERROR: Screen names must be unique", file=sys.stderr)
raise SystemExit(1)


def config_action(args):
update_config(args.config_file, {args.key: args.value})

Expand Down
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

setuptools.setup(
name="31",
version="2.2",
version="2.3.0",
author="Kavi Gupta",
author_email="31@kavigupta.org",
description="Runs code in a specified environment in the background and notifies you when it is done.",
Expand Down
126 changes: 126 additions & 0 deletions tests/duplicate_names_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
import subprocess

from .test31 import Test31, RC_TEST
from s31.main import _check_screen_name_non_overlap


class DuplicateNamesTest(Test31):
def test_check_function_directly(self):
"""Test the duplicate check function directly"""
commands = [
("echo_1", None, None),
("echo_1", None, None),
("echo_2", None, None),
]
with self.assertRaises(SystemExit) as cm:
_check_screen_name_non_overlap(commands)
self.assertEqual(cm.exception.code, 1)
def test_duplicate_values_in_foreach(self):
"""Test that duplicate values in foreach generate duplicate names error"""
# Run command and capture both stdout and stderr
result = subprocess.run(
["31", "c", "--no-email", "-s", "-f", "%x", "1,1,2", "echo %x", "--config-file", RC_TEST],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
# Should exit with code 1
self.assertEqual(result.returncode, 1, f"Expected exit code 1, got {result.returncode}")
# Should contain error message about duplicate screen names
error_output = result.stderr
self.assertIn("Screen names generated multiple times", error_output)
self.assertIn("ERROR: Screen names must be unique", error_output)
# Check that the duplicate name is listed
self.assertIn("'echo_1'", error_output)

def test_duplicate_after_sanitization(self):
"""Test that values that sanitize to the same name are detected as duplicates"""
result = subprocess.run(
["31", "c", "--no-email", "-s", "-f", "%x", "a b,a-b", "echo %x", "--config-file", RC_TEST],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
self.assertEqual(result.returncode, 1)
error_output = result.stderr
# Both "a b" and "a-b" sanitize to "a_b"
self.assertIn("Screen names generated multiple times", error_output)
self.assertIn("ERROR: Screen names must be unique", error_output)
self.assertIn("'echo_a_b'", error_output)

def test_duplicate_with_explicit_screen_name(self):
"""Test that explicit screen names that result in duplicates are detected"""
result = subprocess.run(
[
"31",
"c",
"--no-email",
"-s",
"-f",
"%x",
"1,2",
"-n",
"test",
"echo %x",
"--config-file",
RC_TEST,
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
self.assertEqual(result.returncode, 1)
error_output = result.stderr
# Both commands would have screen name "test"
self.assertIn("Screen names generated multiple times", error_output)
self.assertIn("ERROR: Screen names must be unique", error_output)
self.assertIn("'test'", error_output)

def test_duplicate_with_zipped_foreach(self):
"""Test that zipped foreach can generate duplicates"""
result = subprocess.run(
[
"31",
"c",
"--no-email",
"-s",
"-f2",
"%x",
"%y",
"1,1",
"a,a",
"echo %x %y",
"--config-file",
RC_TEST,
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
self.assertEqual(result.returncode, 1)
error_output = result.stderr
# Both combinations (1,a) would result in the same screen name
self.assertIn("Screen names generated multiple times", error_output)
self.assertIn("ERROR: Screen names must be unique", error_output)

def test_multiple_duplicates(self):
"""Test that multiple duplicate names are all reported"""
result = subprocess.run(
["31", "c", "--no-email", "-s", "-f", "%x", "1,1,2,2,3", "echo %x", "--config-file", RC_TEST],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
)
self.assertEqual(result.returncode, 1)
error_output = result.stderr
# Should report both 'echo_1' and 'echo_2' as duplicates
self.assertIn("Screen names generated multiple times", error_output)
self.assertIn("ERROR: Screen names must be unique", error_output)
self.assertIn("'echo_1'", error_output)
self.assertIn("'echo_2'", error_output)