# -*- coding: utf-8 -*- """ PC-BASIC test.session unit tests for session API (c) 2020--2023 Rob Hagemans This file is released under the GNU GPL version 3 or later. """ import os import io from io import open import unittest from pcbasic import Session from tests.unit.utils import TestCase, run_tests def istypeval(val, refval): """Check agreement in both type and value.""" return isinstance(val, type(refval)) and val == refval class SessionTest(TestCase): """Unit tests for Session.""" tag = u'session' def test_session(self): """Test basic Session API.""" with Session() as s: s.execute('a=1') assert s.evaluate('a+2') == 3. assert s.evaluate('"abc"+"d"') == b'abcd' assert s.evaluate('string$(a+2, "@")') == b'@@@' # string variable s.set_variable('B$', 'abcd') assert s.get_variable('B$') == b'abcd' assert istypeval(s.evaluate('LEN(B$)'), 4) # unset variable assert s.evaluate('C!') == 0. assert istypeval(s.get_variable('D%'), 0) # unset array s.set_variable('A%()', [[0,0,5], [0,0,6]]) assert s.get_variable('A%()') == [ [0, 0, 5, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 6, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] ] assert s.evaluate('A%(0,2)') == 5 assert s.evaluate('A%(1,2)') == 6 assert s.evaluate('A%(1,7)') == 0 assert s.evaluate('FRE(0)') == 60020. assert s.evaluate('CSRLIN') == 1 s.execute('print b$') assert s.evaluate('CSRLIN') == 2 def test_session_convert(self): """Test Session.convert(variable).""" unicode = type(u'') with Session() as s: # omit to_type assert istypeval(s.convert(1, None), 1) # from_type == to_type assert istypeval(s.convert(1, int), 1) # cp437 string to unicode assert s.convert(b'\x00\x01\x9C\x86\xe0', unicode) == u'\x00\u263a\xa3\xe5\u03b1' # unicode to cp437 assert s.convert(u'\x00\u263a\xa3\xe5\u03b1', bytes) == b'\x00\x01\x9C\x86\xe0' # int to bool assert istypeval(s.convert(0, bool), False) assert istypeval(s.convert(-1, bool), True) assert istypeval(s.convert(1, bool), True) # float to bool assert istypeval(s.convert(0., bool), False) assert istypeval(s.convert(-1., bool), True) # bool to int assert istypeval(s.convert(True, int), -1) assert istypeval(s.convert(False, int), 0) # bool to float assert istypeval(s.convert(True, float), -1.) assert istypeval(s.convert(False, float), 0.) assert isinstance(s.convert(True, float), float) # int to float assert istypeval(s.convert(1, float), 1.0) # float to int (floor) assert istypeval(s.convert(1.1, int), 1) assert istypeval(s.convert(-0.1, int), -1) # error with self.assertRaises(ValueError): s.convert(b'1', int) def test_session_getset_variable(self): """Test Session.set_variable and Session.get_variable.""" with Session() as s: s.set_variable(b'A%', 1) assert istypeval(s.get_variable(b'A%'), 1) # bytes or unicode argument s.set_variable(u'A%', 2) assert istypeval(s.get_variable(b'A%'), 2) s.set_variable(u'A%', 3) assert istypeval(s.get_variable(u'A%'), 3) s.set_variable(u'A%', 3) assert istypeval(s.get_variable(b'A%'), 3) # undefined variable assert istypeval(s.get_variable('A!'), 0.) # sigil must be explicit with self.assertRaises(ValueError): s.set_variable('B', 0.) # sigil must be explicit with self.assertRaises(ValueError): s.get_variable('B', 0.) # boolean s.set_variable(b'A%', True) assert istypeval(s.get_variable(b'A%'), -1) # single s.set_variable(b'A!', 1.1) self.assertAlmostEqual(s.get_variable(b'A!'), 1.1, places=6) # double s.set_variable(b'A#', 0.1234567890123) self.assertAlmostEqual(s.get_variable(b'A#'), 0.1234567890123, places=14) # bytes string s.set_variable(b'A$', b'1') assert s.get_variable(b'A$') == b'1' # unicode string value (sterling sign) s.set_variable(b'A$', u'\xc2\xa3') # bytes output, in cp437 assert s.get_variable(b'A$') == b'\x9C' # undefined array assert s.get_variable('A%()') == [] # can't set array to empty with self.assertRaises(ValueError): s.set_variable('ARR2!()', []) def test_session_evaluate(self): """Test Session.evaluate.""" with Session() as s: s.set_variable(b'A!', 1) # variable, implicit sigil allowed. expression can be bytes or unicode assert s.evaluate(b'A') == 1 assert s.evaluate(u'A') == 1 # expression assert s.evaluate(u'A*2+1') == 3.0 # syntax error assert s.evaluate(b'LOG+1') is None def test_session_evaluate_number(self): """Test Session.evaluate starting with a number.""" with Session() as s: assert s.evaluate(b'1+1') == 2 def test_session_bind_file(self): """test Session.bind_file.""" # open file object with open(self.output_path('testfile'), 'wb') as f: with Session() as s: name = s.bind_file(f) # can use name as string assert len(str(name)) <= 12 # write to file s.execute('open "{0}" for output as 1: print#1, "x"'.format(name)) with open(self.output_path('testfile'), 'rb') as f: output = f.read() assert output == b'x\r\n\x1a' # existing file by name with Session() as s: name = s.bind_file(self.output_path('testfile')) # write to file s.execute('open "{0}" for input as 1'.format(name)) s.execute('input#1, a$') assert s.get_variable('A$') == b'x' # create file by name native_name = self.output_path(u'new-test-file') try: os.remove(native_name) except EnvironmentError: pass with Session() as s: name = s.bind_file(native_name, create=True) s.execute('open "{0}" for output as 1: print#1, "test";: close'.format(name)) with open(native_name, 'rb') as f: output = f.read() assert output == b'test\x1a' # existing file by BASIC name with Session(devices={b'Z': self.output_path()}) as s: name = s.bind_file(b'Z:TESTFILE') # write to file s.execute('open "{0}" for input as 1'.format(name)) s.execute('input#1, a$') assert s.get_variable('A$') == b'x' # create file by name, provide BASIC name (bytes) native_name = self.output_path(u'new-test-file').encode('ascii') try: os.remove(native_name) except EnvironmentError: pass with Session() as s: name = s.bind_file(native_name, name=b'A B C', create=True) s.execute(b'open "@:A B C" for output as 1: print#1, "test";: close') with open(native_name, 'rb') as f: output = f.read() assert output == b'test\x1a' # create file by name, provide BASIC name (unicode) native_name = self.output_path(u'new-test-file') try: os.remove(native_name) except EnvironmentError: pass with Session() as s: name = s.bind_file(native_name, name=u'A B C', create=True) s.execute(u'open "@:A B C" for output as 1: print#1, "test";: close') with open(native_name, 'rb') as f: output = f.read() assert output == b'test\x1a' def test_session_greeting(self): """Test welcome screen.""" with Session() as s: s.greet() output = [_row.strip() for _row in self.get_text(s)] assert output[0].startswith(b'PC-BASIC ') assert output[1].startswith(b'(C) Copyright 2013--') assert output[1].endswith(b' Rob Hagemans.') assert output[2] == b'60300 Bytes free' assert output[-1] == ( b'1LIST 2RUN\x1b 3LOAD" 4SAVE" 5CONT\x1b' b' 6,"LPT1 7TRON\x1b 8TROFF\x1b 9KEY 0SCREEN' ) def test_session_press_keys(self): """Test Session.press_keys.""" with Session() as s: # eascii: up, esc, SYSTEM, enter s.press_keys(u'\0\x48\x1bSYSTEM\r') s.interact() # note that SYSTEM raises an exception absorbed by the context manager # no nothing further in this block will be executed output = [_row.strip() for _row in self.get_text(s)] # OK prompt should have been overwritten assert output[0] == b'SYSTEM' def test_session_execute(self): """Test Session.execute.""" with Session() as s: # statement s.execute(b'?LOG(1)') # break s.execute(b'STOP') # error s.execute(b'A') output = [_row.strip() for _row in self.get_text(s)] # \xff checked against DOSbox/GW-BASIC assert output[:3] == [b'0', b'Break\xff', b'Syntax error\xff'] assert output[3:] == [b''] * 22 def test_session_no_streams(self): """Test Session without stream copy.""" with Session(input_streams=None, output_streams=None) as s: s.execute(b'a=1') s.execute(b'print a') output = self.get_text_stripped(s) assert output[:1] == [b' 1'] def test_session_iostreams(self): """Test Session with copy to BytesIO.""" bi = io.BytesIO() with Session(input_streams=None, output_streams=bi) as s: s.execute(b'a=1') s.execute(b'print a') assert bi.getvalue() == b' 1 \r\n' def test_session_inputstr_iostreams(self): """Test Session with INPUT$ reading from pipe.""" bi = io.BytesIO(b'ab\x80cd') with Session(input_streams=bi, output_streams=None) as s: abc = s.evaluate(b'input$(3)') assert abc == b'ab\x80', abc @unittest.skip('correct behaviour as yet undecided.') def test_session_inputstr_iostreams_short(self): """Test Session with INPUT$ reading from pipe.""" bi = io.BytesIO(b'ab') with Session(input_streams=bi, output_streams=None) as s: abc = s.evaluate(b'input$(3)') assert abc == b'ab', abc def test_session_inputstr_iostreams_closed(self): """Test Session with INPUT$ reading from pipe.""" bi = io.BytesIO(b'abc') with Session(input_streams=bi, output_streams=None) as s: abc = s.evaluate(b'input$(3)') assert abc == b'abc', abc def test_session_inputstr_iostreams_unicode(self): """Test Session with INPUT$ reading from pipe.""" bi = io.StringIO(u'ab£cd') with Session(input_streams=bi, output_streams=None) as s: abc = s.evaluate(b'input$(3)') assert abc == b'ab\x9C', abc def test_session_inputstr_iostreams_file(self): """Test Session with INPUT$ reading from pipe.""" with open(self.output_path('testfile'), 'w+b') as f: f.write(b'ab\x80cd') f.seek(0) with Session(input_streams=f, output_streams=None) as s: abc = s.evaluate(b'input$(3)') assert abc == b'ab\x80', abc @unittest.skip('correct behaviour as yet undecided.') def test_session_inputstr_iostreams_file_short(self): """Test Session with INPUT$ reading from pipe.""" with open(self.output_path('testfile'), 'w+b') as f: f.write(b'ab') f.seek(0) with Session(input_streams=f, output_streams=None) as s: abc = s.evaluate(b'input$(3)') assert abc == b'ab', abc def test_session_inputstr_iostreams_unicode_file(self): """Test Session with INPUT$ reading from pipe.""" with open(self.output_path('testfile'), 'w+') as f: f.write(u'ab£cd') f.seek(0) with Session(input_streams=f, output_streams=None) as s: abc = s.evaluate(b'input$(3)') assert abc == b'ab\x9C', abc def test_session_bad_type_iostreams(self): """Test Session with iostreams of incorrect type.""" with self.assertRaises(TypeError): Session(input_streams=1).start() with self.assertRaises(TypeError): Session(output_streams=2).start() def test_session_printcopy(self): """Test Session with ctrl print-screen copy.""" with Session( input_streams=None, output_streams=None, devices={'LPT1': 'FILE:{}'.format(self.output_path('print.txt'))} ) as s: # ctrl+printscreen s.press_keys(u'\0\x72') s.press_keys(u'system\r') s.interact() with open(self.output_path('print.txt'), 'rb') as f: output = f.read() assert output == b'system\r\n', repr(output) def test_session_no_printcopy(self): """Test Session switching off ctrl print-screen copy.""" with Session( input_streams=None, output_streams=None, devices={'LPT1': 'FILE:{}'.format(self.output_path('print.txt'))} ) as s: # ctrl+printscreen s.press_keys(u'\0\x72\0\x72') s.press_keys(u'system\r') s.interact() with open(self.output_path('print.txt')) as f: assert f.read() == '' def test_gosub_from_direct_line(self): """Test for issue#184: GOSUB from direct line should not RETURN into program.""" SOURCE = """\ 10 PRINT "Main" 30 A = -42 40 END 50 PRINT "After End" 60 A = 42 70 RETURN """ with Session() as session: session.execute(SOURCE) session.execute("GOSUB 60") assert session.evaluate('A') == 42 def test_to_list_off_by_one(self): """Test for issue #182: range off by one in to_list.""" with Session() as session: session.execute(""" DIM J2(5,2) J2(1,1)=-1 J2(1,2)=-1 """) assert session.get_variable('J2!()') == [ [0.0, 0.0, 0.0], [0.0, -1.0, -1.0], [0.0, 0.0, 0.0], [0.0, 0.0, 0.0], [0.0, 0.0, 0.0], [0.0, 0.0, 0.0] ] from pcbasic.basic import iostreams from pcbasic.basic.codepage import Codepage class NonBlockingInputWrapperTest(TestCase): """Unit tests for NonBlockingInputWrapper.""" tag = u'nonblockinginputwrapper' def test_read_lfcr(self): """Test read() with LF/CR conversion.""" with io.open(self.output_path('inp.txt'), 'w') as f: f.write(u'12\n34') stream = io.open(self.output_path('inp.txt'), 'r') nbiw = iostreams.NonBlockingInputWrapper(stream, Codepage(), lfcr=True) assert nbiw.read() == u'12\r34' if __name__ == '__main__': run_tests()