Dean
/
Bookdata-tools
mirror of https://github.com/BoiseState/bookdata-tools


  
1

	
2

	
3

	
4

	
5

	
6

	
7

	
8

	
9

	
10

	
11

	
12

	
13

	
14

	
15

	
16

	
17

	
18

	
19

	
20

	
21

	
22

	
23

	
24

	
25

	
26

	
27

	
28

	
29

	
30

	
31

	
32

	
33

	
34

	
35

	
36

	
37

	
38

	
39

	
40

	
41

	
42

	
43

	
44

	
45

	
46

	
47

	
48

	
49

	
50

	
51

	
52

	
53

	
54

	
55

	
56

	
57

	
58

	
59

	
60

	
61

	
62

	
63

	
64

	
65

	
66

	
67

	
68

	
69

	
70

	
71

	
72

	
73

	
74

	
75

	
76

	
77

	
78

	
79

	
80

	
81

	
82

	
            from io import StringIO
from psycopg2 import errorcodes

import support as s


def test_single_query():
    query = 'SELECT * FROM pizza'
    script = s.SqlScript(StringIO(query))
    assert len(script.chunks) == 1
    assert script.chunks[0].allowed_errors == []
    assert script.chunks[0].src == query
    assert script.chunks[0].use_transaction


def test_two_queries():
    query = '''
SELECT * FROM pizza;
---
SELECT fish FROM chips;
'''
    script = s.SqlScript(StringIO(query))
    assert len(script.chunks) == 2
    assert script.chunks[0].src.strip() == 'SELECT * FROM pizza;'
    assert script.chunks[0].allowed_errors == []
    assert script.chunks[1].src.strip() == 'SELECT fish FROM chips;'
    assert script.chunks[1].allowed_errors == []
    assert script.chunks[0].use_transaction


def test_label_query():
    query = '''
--- #step Select from pizza
SELECT * FROM pizza;
'''
    script = s.SqlScript(StringIO(query))
    assert len(script.chunks) == 1
    assert script.chunks[0].src.strip() == 'SELECT * FROM pizza;'
    assert script.chunks[0].allowed_errors == []
    assert script.chunks[0].label == 'Select from pizza'
    assert script.chunks[0].use_transaction


def test_notx_query():
    query = '''
--- #notx
SELECT * FROM pizza;
'''
    script = s.SqlScript(StringIO(query))
    assert len(script.chunks) == 1
    assert script.chunks[0].src.strip() == 'SELECT * FROM pizza;'
    assert script.chunks[0].allowed_errors == []
    assert not script.chunks[0].use_transaction


def test_allow_errors():
    query = '''
--- #allow invalid_object_definition
SELECT * FROM pizza;
'''
    script = s.SqlScript(StringIO(query))
    assert len(script.chunks) == 1
    assert script.chunks[0].src.strip() == 'SELECT * FROM pizza;'
    assert script.chunks[0].allowed_errors == [errorcodes.INVALID_OBJECT_DEFINITION]


def test_allow_multiple():
    query = '''
--- #allow invalid_object_definition
SELECT * FROM pizza;
--- #allow invalid_object_definition
--- #allow duplicate_function
CREATE FUNCTION foobar;
'''
    script = s.SqlScript(StringIO(query))
    assert len(script.chunks) == 2
    assert script.chunks[0].src.strip() == 'SELECT * FROM pizza;'
    assert script.chunks[0].allowed_errors == [errorcodes.INVALID_OBJECT_DEFINITION]

    assert script.chunks[1].src.strip() == 'CREATE FUNCTION foobar;'
    assert script.chunks[1].allowed_errors == \
        [errorcodes.INVALID_OBJECT_DEFINITION, errorcodes.DUPLICATE_FUNCTION]