summaryrefslogtreecommitdiff
path: root/tests/test_functions.py
blob: d5d8f9155df79d00ecf9e9d3c6837bb566ff33e7 (plain)
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
'''
Created on 13/02/2012

@author: piranna
'''
from unittest import main, TestCase

from sqlparse.filters import Tokens2Unicode
from sqlparse.lexer   import tokenize

import sys
sys.path.insert(0, '..')

from sqlparse.filters   import compact
from sqlparse.functions import getcolumns, getlimit, IsType


class Test_SQL(TestCase):
    sql = """-- type: script
            -- return: integer

            INCLUDE "_Make_DirEntry.sql";

            INSERT INTO directories(inode)
                            VALUES(:inode)
            LIMIT 1"""

    sql2 = """SELECT child_entry,asdf AS inode, creation
              FROM links
              WHERE parent_dir == :parent_dir AND name == :name
              LIMIT 1"""

    sql3 = """SELECT
    0 AS st_dev,
    0 AS st_uid,
    0 AS st_gid,

    dir_entries.type         AS st_mode,
    dir_entries.inode        AS st_ino,
    COUNT(links.child_entry) AS st_nlink,

    :creation                AS st_ctime,
    dir_entries.access       AS st_atime,
    dir_entries.modification AS st_mtime,
--    :creation                                                AS st_ctime,
--    CAST(STRFTIME('%s',dir_entries.access)       AS INTEGER) AS st_atime,
--    CAST(STRFTIME('%s',dir_entries.modification) AS INTEGER) AS st_mtime,

    COALESCE(files.size,0) AS st_size, -- Python-FUSE
    COALESCE(files.size,0) AS size     -- PyFilesystem

FROM dir_entries
    LEFT JOIN files
        ON dir_entries.inode == files.inode
    LEFT JOIN links
        ON dir_entries.inode == links.child_entry

WHERE dir_entries.inode == :inode

GROUP BY dir_entries.inode
LIMIT 1"""


class Test_Compact(Test_SQL):
    def test_compact1(self):
        self.assertEqual(Tokens2Unicode(compact(self.sql, 'tests/files')),
            'INSERT INTO dir_entries(type)VALUES(:type);INSERT INTO '
            'directories(inode)VALUES(:inode)LIMIT 1')

    def test_compact2(self):
        self.assertEqual(Tokens2Unicode(compact(self.sql2)),
            'SELECT child_entry,asdf AS inode,creation FROM links WHERE '
            'parent_dir==:parent_dir AND name==:name LIMIT 1')

    def test_compact3(self):
        self.assertEqual(Tokens2Unicode(compact(self.sql3)),
            'SELECT 0 AS st_dev,0 AS st_uid,0 AS st_gid,dir_entries.type AS '
            'st_mode,dir_entries.inode AS st_ino,COUNT(links.child_entry)AS '
            'st_nlink,:creation AS st_ctime,dir_entries.access AS st_atime,'
            'dir_entries.modification AS st_mtime,COALESCE(files.size,0)AS '
            'st_size,COALESCE(files.size,0)AS size FROM dir_entries LEFT JOIN'
            ' files ON dir_entries.inode==files.inode LEFT JOIN links ON '
            'dir_entries.inode==links.child_entry WHERE dir_entries.inode=='
            ':inode GROUP BY dir_entries.inode LIMIT 1')


class Test_GetColumns(Test_SQL):
    def test_getcolumns1(self):
        columns = getcolumns(tokenize(self.sql))
        self.assertEqual(columns, [])

    def test_getcolumns2(self):
        columns = getcolumns(tokenize(self.sql2))
        self.assertEqual(columns, ['child_entry', 'inode', 'creation'])

    def test_getcolumns3(self):
        columns = getcolumns(tokenize(self.sql3))
        self.assertEqual(columns, ['st_dev', 'st_uid', 'st_gid', 'st_mode',
                                   'st_ino', 'st_nlink', 'st_ctime',
                                   'st_atime', 'st_mtime', 'st_size', 'size'])


class Test_GetLimit(Test_SQL):
    def test_getlimit1(self):
        limit = getlimit(tokenize(self.sql))
        self.assertEqual(limit, 1)

    def test_getlimit2(self):
        limit = getlimit(tokenize(self.sql2))
        self.assertEqual(limit, 1)

    def test_getlimit3(self):
        limit = getlimit(tokenize(self.sql3))
        self.assertEqual(limit, 1)


class Test_IsType(Test_SQL):
    def test_istype2(self):
        stream = tokenize(self.sql2)
        self.assertTrue(IsType('SELECT')(stream))

        stream = tokenize(self.sql2)
        self.assertFalse(IsType('INSERT')(stream))


if __name__ == "__main__":
    #import sys;sys.argv = ['', 'Test.testName']
    main()