← Back to branch summary

~ubuntu-branches/ubuntu/raring/software-center/raring-proposed

« back to all changes in this revision

Viewing changes to tests/test_recagent.py

  • Committer: Package Import Robot
  • Author(s): Michael Vogt
  • Date: 2012-10-11 15:33:05 UTC
  • mfrom: (195.1.18 quantal)
  • Revision ID: package-import@ubuntu.com-20121011153305-fm5ln7if3rpzts4n
Tags: 5.4.1.1
https://launchpad.net/bugs/1065481
https://launchpad.net/bugs/1060106
* lp:~mvo/software-center/reinstall-previous-purchase-token-fix:
  - fix reinstall previous purchases that have a system-wide
    license key LP: #1065481
* lp:~mvo/software-center/lp1060106:
  - Add missing gettext init for utils/update-software-center-agent
    (LP: #1060106)

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_recagent.py
10
10
)
11
11
setup_test_env()
12
12
 
 
13
import softwarecenter
13
14
from softwarecenter.backend.recagent import RecommenderAgent
14
15
 
15
16
 
16
 
class TestRecommenderAgent(unittest.TestCase):
 
17
class MockTestRecommenderAgent(unittest.TestCase):
 
18
 
 
19
    @patch.object(softwarecenter.backend.recagent.SpawnHelper, 
 
20
                  'run_generic_piston_helper')
 
21
    def test_mocked_recagent_post_submit_profile(self, mock_spawn_helper_run):
 
22
        recommender_agent = RecommenderAgent()
 
23
        recommender_agent._calc_profile_id = lambda profile: "i-am-random"
 
24
        db = get_test_db()
 
25
        recommender_agent.post_submit_profile(db)
 
26
        args, kwargs =  mock_spawn_helper_run.call_args
 
27
        # ensure we have packages in the package list and the
 
28
        # kwargs have the names we expect
 
29
        self.assertNotEqual(kwargs['data'][0]['package_list'], [])
 
30
 
 
31
 
 
32
class RealTestRecommenderAgent(unittest.TestCase):
17
33
    """ tests the recommender agent """
18
 
 
 
34
    
 
35
    @unittest.skipIf(os.getuid() == 0, 
 
36
                     "this is not supported running as root")
19
37
    def setUp(self):
20
38
        self.loop = GObject.MainLoop(GObject.main_context_default())
21
39
        self.error = False
27
45
            self.addCleanup(os.environ.pop, "SOFTWARE_CENTER_RECOMMENDER_HOST")
28
46
        server = "https://rec.staging.ubuntu.com"
29
47
        os.environ["SOFTWARE_CENTER_RECOMMENDER_HOST"] = server
30
 
 
31
 
    @patch('softwarecenter.backend.recagent.SpawnHelper'
32
 
           '.run_generic_piston_helper')
33
 
    def test_mocked_recagent_post_submit_profile(self, mock_spawn_helper_run):
34
 
        def _patched_on_submit_profile_data(*args, **kwargs):
35
 
            piston_submit_profile = {}
36
 
            recommender_agent.emit("submit-profile-finished",
37
 
                                   piston_submit_profile)
38
 
        mock_spawn_helper_run.side_effect = _patched_on_submit_profile_data
39
 
        recommender_agent = RecommenderAgent()
40
 
        recommender_agent.connect("submit-profile-finished", self.on_query_done)
41
 
        recommender_agent.connect("error", self.on_query_error)
42
 
        recommender_agent._calc_profile_id = lambda profile: "i-am-random"
43
 
        db = get_test_db()
44
 
        recommender_agent.post_submit_profile(db)
45
 
        self.assertFalse(self.error)
46
 
        args, kwargs =  mock_spawn_helper_run.call_args
47
 
        self.assertNotEqual(kwargs['data'][0]['package_list'], [])
 
48
        # most tests need it
 
49
        self.recommender_agent = RecommenderAgent()
 
50
        self.recommender_agent.connect("error", self.on_query_error)
48
51
 
49
52
    def on_query_done(self, recagent, data):
50
 
        # print "query done, data: '%s'" % data
 
53
        #print "query done, data: '%s'" % data
51
54
        self.loop.quit()
 
55
        self.error = False
 
56
        self.error_msg = ""
52
57
 
53
58
    def on_query_error(self, recagent, error):
54
 
        # print "query error received: ", error
 
59
        #print "query error received: ", error
55
60
        self.loop.quit()
56
61
        self.error = True
 
62
        self.error_msg = error
 
63
 
 
64
    def assertServerReturnsWithNoError(self):
 
65
        self.loop.run()
 
66
        self.assertFalse(self.error, "got error: '%s'" % self.error_msg)
57
67
 
58
68
    def test_recagent_query_server_status(self):
59
 
        # NOTE: This requires a working recommender host that is reachable
60
 
        recommender_agent = RecommenderAgent()
61
 
        recommender_agent.connect("server-status", self.on_query_done)
62
 
        recommender_agent.connect("error", self.on_query_error)
63
 
        recommender_agent.query_server_status()
64
 
        self.loop.run()
65
 
        self.assertFalse(self.error)
 
69
        self.recommender_agent.connect("server-status", self.on_query_done)
 
70
        self.recommender_agent.query_server_status()
 
71
        self.assertServerReturnsWithNoError()
66
72
 
67
 
    # FIXME: disabled for now as the server is not quite working
68
 
    def disabled_test_recagent_post_submit_profile(self):
 
73
    @unittest.skip("server returns 401")
 
74
    def test_recagent_post_submit_profile(self):
69
75
        # NOTE: This requires a working recommender host that is reachable
70
 
        recommender_agent = RecommenderAgent()
71
 
        recommender_agent.connect("submit-profile-finished", self.on_query_done)
72
 
        recommender_agent.connect("error", self.on_query_error)
73
76
        db = get_test_db()
74
 
        recommender_agent.post_submit_profile(db)
75
 
        self.loop.run()
76
 
        self.assertFalse(self.error)
 
77
        self.recommender_agent.connect(
 
78
            "submit-profile-finished", self.on_query_done)
 
79
        self.recommender_agent.post_submit_profile(db)
 
80
        self.assertServerReturnsWithNoError()
77
81
        #print mock_request._post
78
 
 
79
 
     # NOTE: this server call is currently not needed and not used
80
 
#    def disabled_test_recagent_query_submit_anon_profile(self):
81
 
#        # NOTE: This requires a working recommender host that is reachable
82
 
#        recommender_agent = RecommenderAgent()
83
 
#        recommender_agent.connect("submit-anon-profile", self.on_query_done)
84
 
#        recommender_agent.connect("error", self.on_query_error)
85
 
#        recommender_agent.query_submit_anon_profile(
86
 
#                uuid=recommender_uuid,
87
 
#                installed_packages=["pitivi", "fretsonfire"],
88
 
#                extra="")
89
 
#        self.loop.run()
90
 
#        self.assertFalse(self.error)
91
 
 
92
 
    # FIXME: disabled for now as the server is not quite working
93
 
    def disabled_test_recagent_query_profile(self):
94
 
        # NOTE: This requires a working recommender host that is reachable
95
 
        recommender_agent = RecommenderAgent()
96
 
        recommender_agent.connect("profile", self.on_query_done)
97
 
        recommender_agent.connect("error", self.on_query_error)
98
 
        recommender_agent.query_profile(pkgnames=["pitivi", "fretsonfire"])
99
 
        self.loop.run()
100
 
        self.assertFalse(self.error)
101
 
 
102
 
    # FIXME: disabled for now as the server is not quite working
103
 
    def disabled_test_recagent_query_recommend_me(self):
104
 
        # NOTE: This requires a working recommender host that is reachable
105
 
        recommender_agent = RecommenderAgent()
106
 
        recommender_agent.connect("recommend-me", self.on_query_done)
107
 
        recommender_agent.connect("error", self.on_query_error)
108
 
        recommender_agent.query_recommend_me()
109
 
        self.loop.run()
110
 
        self.assertFalse(self.error)
 
82
        
 
83
    @unittest.skip("server returns 401")
 
84
    def test_recagent_query_submit_anon_profile(self):
 
85
        self.recommender_agent.connect(
 
86
            "submit-anon-profile-finished", self.on_query_done)
 
87
        self.recommender_agent.post_submit_anon_profile(
 
88
                uuid="xxxyyyzzz",
 
89
                installed_packages=["pitivi", "fretsonfire"],
 
90
                extra="")
 
91
        self.assertServerReturnsWithNoError()
 
92
 
 
93
    @unittest.skip("server returns 401")
 
94
    def test_recagent_query_profile(self):
 
95
        self.recommender_agent.connect("profile", self.on_query_done)
 
96
        self.recommender_agent.query_profile(pkgnames=["pitivi", "fretsonfire"])
 
97
        self.assertServerReturnsWithNoError()
 
98
 
 
99
    @unittest.skip("server returns 401")
 
100
    def test_recagent_query_recommend_me(self):
 
101
        self.recommender_agent.connect("recommend-me", self.on_query_done)
 
102
        self.recommender_agent.query_recommend_me()
 
103
        self.assertServerReturnsWithNoError()
111
104
 
112
105
    def test_recagent_query_recommend_app(self):
113
 
        # NOTE: This requires a working recommender host that is reachable
114
 
        recommender_agent = RecommenderAgent()
115
 
        recommender_agent.connect("recommend-app", self.on_query_done)
116
 
        recommender_agent.connect("error", self.on_query_error)
117
 
        recommender_agent.query_recommend_app("pitivi")
118
 
        self.loop.run()
119
 
        self.assertFalse(self.error)
 
106
        self.recommender_agent.connect("recommend-app", self.on_query_done)
 
107
        self.recommender_agent.query_recommend_app("pitivi")
 
108
        self.assertServerReturnsWithNoError()
120
109
 
121
 
    # disabled for now (2012-03-20) as the server is returning 504
122
 
    def disabled_test_recagent_query_recommend_all_apps(self):
123
 
        # NOTE: This requires a working recommender host that is reachable
124
 
        recommender_agent = RecommenderAgent()
125
 
        recommender_agent.connect("recommend-all-apps", self.on_query_done)
126
 
        recommender_agent.connect("error", self.on_query_error)
127
 
        recommender_agent.query_recommend_all_apps()
128
 
        self.loop.run()
129
 
        self.assertFalse(self.error)
 
110
    def test_recagent_query_recommend_all_apps(self):
 
111
        self.recommender_agent.connect("recommend-all-apps", self.on_query_done)
 
112
        self.recommender_agent.query_recommend_all_apps()
 
113
        self.assertServerReturnsWithNoError()
130
114
 
131
115
    def test_recagent_query_recommend_top(self):
132
 
        # NOTE: This requires a working recommender host that is reachable
133
 
        recommender_agent = RecommenderAgent()
134
 
        recommender_agent.connect("recommend-top", self.on_query_done)
135
 
        recommender_agent.connect("error", self.on_query_error)
136
 
        recommender_agent.query_recommend_top()
137
 
        self.loop.run()
138
 
        self.assertFalse(self.error)
 
116
        self.recommender_agent.connect("recommend-top", self.on_query_done)
 
117
        self.recommender_agent.query_recommend_top()
 
118
        self.assertServerReturnsWithNoError()
139
119
 
140
 
    @unittest.skip('Disabled, should be re-enabled ASAP')
141
120
    def test_recagent_query_error(self):
142
121
        # NOTE: This tests the error condition itself! it simply forces an error
143
122
        #       'cuz there definitely isn't a server here  :)
150
129
        self.loop.run()
151
130
        self.assertTrue(self.error)
152
131
 
 
132
    @unittest.skip("server returns 401")
 
133
    def test_recagent_post_implicit_feedback(self):
 
134
        self.recommender_agent.connect("submit-implicit-feedback-finished",
 
135
                                  self.on_query_done)
 
136
        from softwarecenter.enums import RecommenderFeedbackActions
 
137
        self.recommender_agent.post_implicit_feedback(
 
138
                "bluefish",
 
139
                RecommenderFeedbackActions.INSTALLED)
 
140
        self.assertServerReturnsWithNoError()
 
141
 
 
142
 
153
143
if __name__ == "__main__":
 
144
    #import logging
 
145
    #logging.basicConfig(level=logging.DEBUG)
154
146
    unittest.main()