tests/test_examples.py

import importlib.util
from pathlib import Path
import tempfile
import unittest

ROOT=Path(__file__).resolve().parents[1]

def load(rel,name):
    spec=importlib.util.spec_from_file_location(name,ROOT/rel)
    obj=importlib.util.module_from_spec(spec)
    spec.loader.exec_module(obj)
    return obj

csvtool=load('examples/select_csv_columns.py','csvtool')
inv=load('examples/safe_file_inventory.py','inv')

class CSVExamples(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.root=Path(self.tmp.name)
        self.src=self.root/'input.csv'; self.dst=self.root/'output.csv'
    def test_multiline_quoted_fields(self):
        self.src.write_text('id,notes,x\n001,"one,two\nthree",y\n',encoding='utf-8')
        self.assertEqual(csvtool.select_columns(self.src,self.dst,['id','notes']),1)
        self.assertEqual(self.dst.read_text(encoding='utf-8'),'id,notes\n001,"one,two\nthree"\n')
    def test_leading_zeros(self):
        self.src.write_text('code,value\n00007,5\n',encoding='utf-8')
        csvtool.select_columns(self.src,self.dst,['code'])
        self.assertIn('00007',self.dst.read_text())
    def test_missing_column_rejected(self):
        self.src.write_text('a,b\n1,2\n',encoding='utf-8')
        with self.assertRaises(ValueError):csvtool.select_columns(self.src,self.dst,['missing'])
        self.assertFalse(self.dst.exists())
    def test_duplicate_input_header_rejected(self):
        self.src.write_text('a,a\n1,2\n',encoding='utf-8')
        with self.assertRaises(ValueError):csvtool.select_columns(self.src,self.dst,['a'])
    def test_in_place_rejected(self):
        self.src.write_text('a,b\n1,2\n',encoding='utf-8')
        with self.assertRaises(ValueError):csvtool.select_columns(self.src,self.src,['a'])
    def test_existing_output_rejected(self):
        self.src.write_text('a,b\n1,2\n',encoding='utf-8')
        self.dst.write_text('keep',encoding='utf-8')
        with self.assertRaises(FileExistsError):csvtool.select_columns(self.src,self.dst,['a'])
        self.assertEqual(self.dst.read_text(),'keep')
    def test_failure_leaves_no_partial_output(self):
        self.src.write_text('a,b\n1,2,3\n',encoding='utf-8')
        with self.assertRaises(ValueError):csvtool.select_columns(self.src,self.dst,['a'])
        self.assertFalse(self.dst.exists())
        self.assertFalse(list(self.root.glob('*.tmp')))
    def test_ragged_row_rejected(self):
        self.src.write_text('a,b\n1,2,3\n',encoding='utf-8')
        with self.assertRaises(ValueError):csvtool.select_columns(self.src,self.dst,['a'])

class InventoryExamples(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.root=Path(self.tmp.name)
    def test_newline_name(self):
        p=self.root/'line\nbreak.log';p.write_text('ok')
        items=inv.inventory(self.root,'.log')
        self.assertEqual(items[0]['relative_path'],'line\nbreak.log')
    def test_leading_dash(self):
        (self.root/'-danger.log').write_text('ok')
        self.assertEqual(inv.inventory(self.root,'.log')[0]['relative_path'],'-danger.log')
    def test_filter(self):
        (self.root/'a.log').write_text('a')
        (self.root/'b.txt').write_text('b')
        self.assertEqual(len(inv.inventory(self.root,'.log')),1)
    def test_does_not_follow_symlinks(self):
        p=self.root/'file.log';p.write_text('safe')
        (self.root/'link.log').symlink_to(p)
        self.assertEqual([x['relative_path'] for x in inv.inventory(self.root,'.log')],['file.log'])
    def test_empty(self):self.assertEqual(inv.inventory(self.root,''),[])

if __name__=='__main__':unittest.main()