tests: add tests for json data source

tests/test_datasource.py

@@ -1,3 +1,4 @@
+import json
 import sys
 import shutil
 
@@ -5,7 +6,7 @@ from quixote import cleanup
 from wcs import publisher
 from wcs.qommon.http_request import HTTPRequest
 from wcs.qommon.form import *
-from wcs import fields
+from wcs import fields, data_sources
 
 from test_widgets import MockHtmlForm, mock_form_submission
 from utilities import create_temporary_pub
@@ -49,3 +50,47 @@ def test_item_field_python_datasource():
     form = MockHtmlForm(widget)
     mock_form_submission(req, widget)
     assert widget.parse() == '1'
+
+def test_python_datasource():
+    plain_list = repr([('1', 'foo'), ('2', 'bar')])
+    datasource = {'type': 'formula', 'value': repr(plain_list)}
+    assert data_sources.get_items(datasource) == plain_list
+
+def test_json_datasource():
+    datasource = {'type': 'json', 'value': ''}
+    assert data_sources.get_items(datasource) == []
+
+    # missing file
+    json_file_path = os.path.join(pub.app_dir, 'test.json')
+    datasource = {'type': 'json', 'value': 'file://%s' % json_file_path}
+    assert data_sources.get_items(datasource) == []
+
+    # invalid json file
+    json_file = open(json_file_path, 'w')
+    json_file.write(u'foobar'.encode('zlib_codec'))
+    json_file.close()
+    assert data_sources.get_items(datasource) == []
+
+    # empty json file
+    json_file = open(json_file_path, 'w')
+    json.dump({}, json_file)
+    json_file.close()
+    assert data_sources.get_items(datasource) == []
+
+    # unrelated json file
+    json_file = open(json_file_path, 'w')
+    json.dump('foobar', json_file)
+    json_file.close()
+    assert data_sources.get_items(datasource) == []
+
+    # another unrelated json file
+    json_file = open(json_file_path, 'w')
+    json.dump({'data': 'foobar'}, json_file)
+    json_file.close()
+    assert data_sources.get_items(datasource) == []
+
+    # a good json file
+    json_file = open(json_file_path, 'w')
+    json.dump({'data': [{'id': '1', 'text': 'foo'}, {'id': '2', 'text': 'bar'}]}, json_file)
+    json_file.close()
+    assert data_sources.get_items(datasource) == [('1', 'foo', '1'), ('2', 'bar', '2')]

wcs/data_sources.py

@@ -100,7 +100,12 @@ def get_items(data_source):
         charset = get_publisher().site_charset
         try:
             results = []
-            for entry in json.load(urllib2.urlopen(url)).get('data'):
+            entries = json.load(urllib2.urlopen(url))
+            if type(entries) is not dict:
+                raise ValueError('not a json dict')
+            if type(entries.get('data')) is not list:
+                raise ValueError('not a json dict with a data list attribute')
+            for entry in entries.get('data'):
                 id = entry.get('id')
                 text = entry.get('text')
                 if type(id) is unicode: