odoo
/
odoo
2
0
Fork 0
Code Pull Requests Releases Activity

[FIX] osv.query: improve support of chained explicit JOINs + corresponding tests 

bzr revid: odo@openerp.com-20101001143642-z3ahichcj55csxib
This commit is contained in:
Olivier Dony 2010-10-01 16:36:42 +02:00
parent 109f51e125
commit 9454f780af
2 changed files with 38 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
bin/osv/query.py
View File

@ -99,13 +99,19 @@ class Query(object):
"""Returns (query_from, query_where, query_params)"""
query_from = ''
tables_to_process = list(self.tables)
def add_joins_for_table(table, query_from):
for (dest_table, lhs_col, col, join) in self.joins.get(table,[]):
tables_to_process.remove(dest_table)
query_from += ' %s %s ON (%s."%s" = %s."%s")' % \
(join, dest_table, table, lhs_col, dest_table, col)
query_from = add_joins_for_table(dest_table, query_from)
return query_from
for table in tables_to_process:
query_from += ' %s ' % table
query_from += table
if table in self.joins:
for (dest_table, lhs_col, col, join) in self.joins[table]:
tables_to_process.remove(dest_table)
query_from += '%s %s ON (%s."%s" = %s."%s")' % \
(join, dest_table, table, lhs_col, dest_table, col)
query_from = add_joins_for_table(table, query_from)
query_from += ','
query_from = query_from[:-1] # drop last comma
return (query_from, " AND ".join(self.where_clause), self.where_clause_params)

30
bin/test/test_osv.py
View File

@ -31,10 +31,34 @@ class QueryTestCase(unittest.TestCase):
query.join(("product_template", "product_category", "categ_id", "id"), outer=False) # add normal join
query.join(("product_product", "res_user", "user_id", "id"), outer=True) # outer join
self.assertEquals(query.get_sql()[0].strip(),
""""product_product" LEFT JOIN "res_user" ON ("product_product"."user_id" = "res_user"."id"), "product_template" JOIN "product_category" ON ("product_template"."categ_id" = "product_category"."id") """.strip(), "Incorrect query")
self.assertEquals(query.get_sql()[1].strip(), """product_product.template_id = product_template.id""".strip(), "Incorrect where clause")
""""product_product" LEFT JOIN "res_user" ON ("product_product"."user_id" = "res_user"."id"),"product_template" JOIN "product_category" ON ("product_template"."categ_id" = "product_category"."id") """.strip())
self.assertEquals(query.get_sql()[1].strip(), """product_product.template_id = product_template.id""".strip())
def test_query_chained_explicit_joins(self):
query = Query()
query.tables.extend(['"product_product"','"product_template"'])
query.where_clause.append("product_product.template_id = product_template.id")
query.join(("product_template", "product_category", "categ_id", "id"), outer=False) # add normal join
query.join(("product_category", "res_user", "user_id", "id"), outer=True) # CHAINED outer join
self.assertEquals(query.get_sql()[0].strip(),
""""product_product","product_template" JOIN "product_category" ON ("product_template"."categ_id" = "product_category"."id") LEFT JOIN "res_user" ON ("product_category"."user_id" = "res_user"."id")""".strip())
self.assertEquals(query.get_sql()[1].strip(), """product_product.template_id = product_template.id""".strip())
def test_mixed_query_chained_explicit_implicit_joins(self):
query = Query()
query.tables.extend(['"product_product"','"product_template"'])
query.where_clause.append("product_product.template_id = product_template.id")
query.join(("product_template", "product_category", "categ_id", "id"), outer=False) # add normal join
query.join(("product_category", "res_user", "user_id", "id"), outer=True) # CHAINED outer join
query.tables.append('"account.account"')
query.where_clause.append("product_category.expense_account_id = account_account.id") # additional implicit join
self.assertEquals(query.get_sql()[0].strip(),
""""product_product","product_template" JOIN "product_category" ON ("product_template"."categ_id" = "product_category"."id") LEFT JOIN "res_user" ON ("product_category"."user_id" = "res_user"."id"),"account.account" """.strip())
self.assertEquals(query.get_sql()[1].strip(), """product_product.template_id = product_template.id AND product_category.expense_account_id = account_account.id""".strip())
def test_raise_missing_lhs(self):
query = Query()
query.tables.append('"product_product"')
self.assertRaises(AssertionError, query.join, ("product_template", "product_category", "categ_id", "id"), outer=False)
self.assertRaises(AssertionError, query.join, ("product_template", "product_category", "categ_id", "id"), outer=False)