structure ajustment

tests/app/__init__.py

@@ -0,0 +1,15 @@
+'''
+Copyright (c) 2013 Qin Xuye <qin@qinxuye.me>
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+'''

tests/app/test_weibo.py

@@ -0,0 +1,156 @@
+'''
+Copyright (c) 2013 Qin Xuye <qin@qinxuye.me>
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+Created on 2013-6-10
+
+@author: Chine
+'''
+import unittest
+import time
+
+from cola.core.opener import MechanizeOpener
+from cola.core.unit import Bundle
+
+from app.weibo import login_hook
+from app.weibo.parsers import MicroBlogParser, ForwardCommentLikeParser, \
+                                    UserInfoParser, UserFriendParser
+from app.weibo.conf import user_config
+from app.weibo.bundle import WeiboUserBundle
+
+from pymongo import Connection
+
+class Test(unittest.TestCase):
+
+
+    def setUp(self):
+        self.test_uid = '1784725941'
+        self.bundle = WeiboUserBundle(self.test_uid)
+        self.opener = MechanizeOpener()
+
+        self.conn = Connection()
+        self.db = self.conn[getattr(user_config.job, 'db')]
+        self.users_collection = self.db.weibo_user
+        self.weibos_collection = self.db.micro_blog
+
+        assert len(user_config.job['login']) > 0
+
+        login_hook(self.opener, **user_config.job['login'][0])
+
+    def _get_urls_and_bundles(self, yields):
+        urls = []
+        bundles = []
+        for item in yields:
+            if isinstance(item, Bundle):
+                bundles.append(item)
+            else:
+                urls.append(item)
+        return urls, bundles
+
+    def tearDown(self):
+        self.users_collection.remove({'uid': self.test_uid})
+        self.weibos_collection.remove({'uid': self.test_uid})
+        self.conn.close()
+
+    def testMicroBlogParser(self):
+        test_url = 'http://weibo.com/aj/mblog/mbloglist?uid=%s&_k=%s' % (
+            self.test_uid,
+            int(time.time() * (10**6))
+        )
+        parser = MicroBlogParser(opener=self.opener, 
+                                 url=test_url, 
+                                 bundle=self.bundle)
+        _, bundles = self._get_urls_and_bundles(parser.parse())
+
+        self.assertEqual(len(bundles), 0)
+
+        size = self.weibos_collection.find({'uid': self.test_uid}).count()
+        self.assertAlmostEqual(size, 15, delta=1)
+
+    def testMicroBlogForwardsParser(self):
+        test_url = 'http://weibo.com/aj/mblog/info/big?id=3596988739933218&_t=0&__rnd=1373094212593'
+        parser = ForwardCommentLikeParser(opener=self.opener,
+                                          url=test_url,
+                                          bundle=self.bundle)
+        urls, _ = self._get_urls_and_bundles(parser.parse())
+
+        self.assertEqual(len(urls), 1)
+
+        weibo = self.weibos_collection.find_one({'mid': '3596988739933218', 'uid': self.test_uid})
+        self.assertLessEqual(len(weibo['forwards']), 20)
+        self.assertGreater(len(weibo['forwards']), 0)
+
+        parser.parse(urls[0])
+        weibo = self.weibos_collection.find_one({'mid': '3596988739933218', 'uid': self.test_uid})
+        self.assertLessEqual(len(weibo['forwards']), 40)
+        self.assertGreater(len(weibo['forwards']), 10)
+        self.assertNotEqual(weibo['forwards'][0], 
+                            weibo['forwards'][10])
+
+    # def testMicroBlogForwardTimeParser(self):
+    #     test_url = 'http://weibo.com/aj/mblog/info/big?id=3596988739933218&_t=0&__rnd=1373094212593'
+    #     parser = ForwardCommentLikeParser(opener=self.opener,
+    #                                       url=test_url,
+    #                                       bundle=self.bundle)
+    #     parser.parse()
+    #
+    #     weibo = self.weibos_collection.find_one({'mid': '3596988739933218', 'uid': self.test_uid})
+    #     self.assertGreater(len(weibo['forwards']), 0)
+
+    def testMicroBlogLikesParser(self):
+        test_url = 'http://weibo.com/aj/like/big?mid=3599246068109415&_t=0&__rnd=1373634556882'
+        parser = ForwardCommentLikeParser(opener=self.opener,
+                                          url=test_url,
+                                          bundle=self.bundle)
+        urls, _ = self._get_urls_and_bundles(parser.parse())
+
+        self.assertEqual(len(urls), 1)
+
+        weibo = self.weibos_collection.find_one({'mid': '3599246068109415', 'uid': self.test_uid})
+        self.assertEqual(len(weibo['likes']), 30)
+
+    def testUserInfoParser(self):
+        test_url = 'http://weibo.com/%s/info' % self.test_uid
+        parser = UserInfoParser(opener=self.opener,
+                                url=test_url,
+                                bundle=self.bundle)
+        parser.parse()
+
+        user = self.users_collection.find_one({'uid': self.test_uid})
+        self.assertTrue('info' in user)
+
+    def testUserInfoParserForSite(self):
+        test_uid = '2733272463'
+        test_url = 'http://weibo.com/%s/info' % test_uid
+        bundle = WeiboUserBundle(test_uid)
+        parser = UserInfoParser(opener=self.opener,
+                                url=test_url,
+                                bundle=bundle)
+        parser.parse()
+
+    def testFriendParser(self):
+        test_url = 'http://weibo.com/%s/follow' % self.test_uid
+        parser = UserFriendParser(opener=self.opener,
+                                  url=test_url,
+                                  bundle=self.bundle)
+        urls, bundles = self._get_urls_and_bundles(parser.parse())
+        self.assertEqual(len(urls), 1)
+        self.assertGreater(bundles, 0)
+
+        user = self.users_collection.find_one({'uid': self.test_uid})
+        self.assertEqual(len(bundles), len(user['follows']))
+
+if __name__ == "__main__":
+    #import sys;sys.argv = ['', 'Test.testParser']
+    unittest.main()

tests/app/test_wiki.py

@@ -0,0 +1,61 @@
+'''
+Copyright (c) 2013 Qin Xuye <qin@qinxuye.me>
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+Created on 2013-5-29
+
+@author: Chine
+'''
+import unittest
+from datetime import datetime
+
+from app.wiki import WikiParser, url_patterns, \
+                         mongo_host, mongo_port, db_name
+
+class FakeWikiParser(WikiParser):
+    def store(self, title, content, last_update):
+        self.title, self.content, self.last_update = title, content, last_update
+
+class Test(unittest.TestCase):
+
+
+    def testWikiParser(self):
+        parser = FakeWikiParser()
+        print dir(parser)
+
+        for url in ('http://en.wikipedia.org/wiki/Python',
+                    'http://zh.wikipedia.org/wiki/Python'):
+            parser.parse(url)
+            lang = url.strip('http://').split('.', 1)[0]
+            self.assertEqual(parser.title, 'Python '+lang)
+            self.assertGreater(len(parser.content), 0)
+            self.assertTrue(isinstance(parser.last_update, datetime))
+
+            self.assertIsNotNone(url_patterns.get_parser(url))
+
+        parser = WikiParser()
+        url = 'http://en.wikipedia.org/wiki/Python'
+        parser.parse(url)
+
+        from pymongo import Connection
+        conn = Connection(mongo_host, mongo_port)
+        db = getattr(conn, db_name)
+        wiki = db.wiki_document.find_one({'title': 'Python en'})
+        self.assertIsNotNone(wiki)
+
+        db.wiki_document.remove({'title': 'Python en'})
+
+if __name__ == "__main__":
+    #import sys;sys.argv = ['', 'Test.testName']
+    unittest.main()