Skip to content

Commit

Permalink
Fixed broken unit tests in 'test_sqlcompletion.py' and made non-class…
Browse files Browse the repository at this point in the history
… type unit tests into class type (dbcli#261)
  • Loading branch information
Gene Lee authored Aug 7, 2019
1 parent 34278ab commit bb5c459
Show file tree
Hide file tree
Showing 5 changed files with 987 additions and 944 deletions.
12 changes: 6 additions & 6 deletions mssqlcli/jsonrpc/contracts/tests/test_json_rpc_contracts.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ def test_successful_connection_AdventureWorks2014(self):
Verify a successful connection response
"""

with open(self.get_test_baseline(
with open(self.get_baseline(\
u'test_simple_query.txt'),
u'r+b',
buffering=0) as response_file:
Expand Down Expand Up @@ -50,7 +50,7 @@ def test_query_execute_response_AdventureWorks2014(self):
"""
Verify a successful query execute response for "select * from HumanResources.Department"
"""
with open(self.get_test_baseline(
with open(self.get_baseline(\
u'test_simple_query.txt'),
u'r+b',
buffering=0) as response_file:
Expand Down Expand Up @@ -78,7 +78,7 @@ def test_query_subset_response_AdventureWorks2014(self):
"""
Test the retrieval of the actual rows for "select * from HumanResources.Department"
"""
with open(self.get_test_baseline(
with open(self.get_baseline(\
u'test_simple_query.txt'),
u'r+b',
buffering=0) as response_file:
Expand Down Expand Up @@ -106,7 +106,7 @@ def test_query_retrieve_correct_response(self):
"""
Verify a query execute request never retrieves query request responses for a different query.
"""
with open(self.get_test_baseline(
with open(self.get_baseline(\
u'test_query_retrieve_correct_response.txt'),
u'r+b',
buffering=0) as response_file:
Expand Down Expand Up @@ -135,7 +135,7 @@ def test_malformed_query_AdventureWorks2014(self):
"""
Verify a failed query execute response for "select * from [HumanResources.Department"
"""
with open(self.get_test_baseline(
with open(self.get_baseline(\
u'test_malformed_query.txt'),
u'r+b',
buffering=0) as response_file:
Expand Down Expand Up @@ -231,7 +231,7 @@ def verify_query_service_response(self,
self.assertEqual(query_row_count, expected_row_count)
self.assertEqual(query_error_events, expected_error_count)

def get_test_baseline(self, file_name):
def get_baseline(self, file_name):
"""
Helper method to get baseline file.
"""
Expand Down
18 changes: 9 additions & 9 deletions mssqlcli/packages/parseutils/tables.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,15 +35,15 @@ def extract_from_part(parsed, stop_at_punctuation=True):
for x in extract_from_part(item, stop_at_punctuation):
yield x
elif stop_at_punctuation and item.ttype is Punctuation:
raise StopIteration
# An incomplete nested select won't be recognized correctly as a
# sub-select. eg: 'SELECT * FROM (SELECT id FROM user'. This causes
# the second FROM to trigger this elif condition resulting in a
# StopIteration. So we need to ignore the keyword if the keyword
# FROM.
# Also 'SELECT * FROM abc JOIN def' will trigger this elif
# condition. So we need to ignore the keyword JOIN and its variants
# INNER JOIN, FULL OUTER JOIN, etc.
# An incomplete nested select won't be recognized correctly as a
# sub-select. eg: 'SELECT * FROM (SELECT id FROM user'. This causes
# the second FROM to trigger this elif condition resulting in a
# StopIteration. So we need to ignore the keyword if the keyword
# FROM.
# Also 'SELECT * FROM abc JOIN def' will trigger this elif
# condition. So we need to ignore the keyword JOIN and its variants
# INNER JOIN, FULL OUTER JOIN, etc.
return
elif item.ttype is Keyword and (
not item.value.upper() == 'FROM') and (
not item.value.upper().endswith('JOIN')):
Expand Down
32 changes: 17 additions & 15 deletions tests/test_prioritization.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,22 @@
from mssqlcli.packages.prioritization import PrevalenceCounter
import unittest

class PrioritizationTests(unittest.TestCase):

def test_prevalence_counter():
counter = PrevalenceCounter()
sql = '''SELECT * FROM foo WHERE bar GROUP BY baz;
select * from foo;
SELECT * FROM foo WHERE bar GROUP
BY baz'''
counter.update(sql)
def test_prevalence_counter(self):
counter = PrevalenceCounter()
sql = '''SELECT * FROM foo WHERE bar GROUP BY baz;
select * from foo;
SELECT * FROM foo WHERE bar GROUP
BY baz'''
counter.update(sql)

keywords = ['SELECT', 'FROM', 'GROUP BY']
expected = [3, 3, 2]
kw_counts = [counter.keyword_count(x) for x in keywords]
assert kw_counts == expected
assert counter.keyword_count('NOSUCHKEYWORD') == 0
keywords = ['SELECT', 'FROM', 'GROUP BY']
expected = [3, 3, 2]
kw_counts = [counter.keyword_count(x) for x in keywords]
assert kw_counts == expected
assert counter.keyword_count('NOSUCHKEYWORD') == 0

names = ['foo', 'bar', 'baz']
name_counts = [counter.name_count(x) for x in names]
assert name_counts == [3, 2, 2]
names = ['foo', 'bar', 'baz']
name_counts = [counter.name_count(x) for x in names]
assert name_counts == [3, 2, 2]
94 changes: 41 additions & 53 deletions tests/test_rowlimit.py
Original file line number Diff line number Diff line change
@@ -1,56 +1,44 @@
import unittest
from mssqlcli.mssql_cli import MssqlCli
from mssqltestutils import create_mssql_cli_options


DEFAULT_OPTIONS = create_mssql_cli_options()
DEFAULT = MssqlCli(DEFAULT_OPTIONS).row_limit
LIMIT = DEFAULT + 1000

low_count = 1
over_default = DEFAULT + 1
over_limit = LIMIT + 1


def test_default_row_limit():
cli = MssqlCli(DEFAULT_OPTIONS)
stmt = "SELECT * FROM students"
result = cli._should_show_limit_prompt(stmt, ['row']*low_count)
assert result is False

result = cli._should_show_limit_prompt(stmt, ['row']*over_default)
assert result is True


def test_set_row_limit():
cli_options = create_mssql_cli_options(row_limit=LIMIT)
cli = MssqlCli(cli_options)
stmt = "SELECT * FROM students"
result = cli._should_show_limit_prompt(stmt, ['row']*over_default)
assert result is False

result = cli._should_show_limit_prompt(stmt, ['row']*over_limit)
assert result is True


def test_no_limit():
cli_options = create_mssql_cli_options(row_limit=0)
cli = MssqlCli(cli_options)
assert cli.row_limit is 0
stmt = "SELECT * FROM students"

result = cli._should_show_limit_prompt(stmt, ['row']*over_limit)
assert result is False


def test_row_limit_on_non_select():
cli = MssqlCli(DEFAULT_OPTIONS)
stmt = "UPDATE students set name='Boby'"
result = cli._should_show_limit_prompt(stmt, None)
assert result is False

cli_options = create_mssql_cli_options(row_limit=0)
assert cli_options.row_limit is 0
cli = MssqlCli(cli_options)
result = cli._should_show_limit_prompt(stmt, ['row']*over_default)
assert cli.row_limit is 0
assert result is False
class RowLimitTests(unittest.TestCase):

DEFAULT_OPTIONS = create_mssql_cli_options()
DEFAULT = MssqlCli(DEFAULT_OPTIONS).row_limit
LIMIT = DEFAULT + 1000

low_count = 1
over_default = DEFAULT + 1
over_limit = LIMIT + 1

def test_default_row_limit(self):
cli = MssqlCli(self.DEFAULT_OPTIONS)
stmt = "SELECT * FROM students"
result = cli._should_show_limit_prompt(stmt, ['row']*self.low_count)
assert result is False

result = cli._should_show_limit_prompt(stmt, ['row']*self.over_default)
assert result is True

def test_no_limit(self):
cli_options = create_mssql_cli_options(row_limit=0)
cli = MssqlCli(cli_options)
assert cli.row_limit is 0
stmt = "SELECT * FROM students"

result = cli._should_show_limit_prompt(stmt, ['row']*self.over_limit)
assert result is False

def test_row_limit_on_non_select(self):
cli = MssqlCli(self.DEFAULT_OPTIONS)
stmt = "UPDATE students set name='Boby'"
result = cli._should_show_limit_prompt(stmt, None)
assert result is False

cli_options = create_mssql_cli_options(row_limit=0)
assert cli_options.row_limit is 0
cli = MssqlCli(cli_options)
result = cli._should_show_limit_prompt(stmt, ['row']*self.over_default)
assert cli.row_limit is 0
assert result is False
Loading

0 comments on commit bb5c459

Please sign in to comment.