def test_timeline_count(self):
        """Test the count parameter of home_timeline"""
        self.app.config['API2_TIMELINE_MAX_RESULTS'] = 50
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            for i in range(60):
                status(project=p, user=u, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(len(data), 20)

        # Test with an acceptable count
        response = self.client.get(self._url(dict(count=50)))
        data = json.loads(response.data)
        eq_(len(data), 50)

        # Test with a count that is too large
        response = self.client.get(self._url(dict(count=60)))
        eq_(response.status_code, 400)

        # Test with a count that is too small
        response = self.client.get(self._url(dict(count=0)))
        eq_(response.status_code, 400)

        # Test with an invalid count
        response = self.client.get(self._url(dict(count='a')))
        eq_(response.status_code, 400)

    def test_timeline_since_id(self):
        """Test the since_id parameter of home_timeline"""
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            for i in range(30):
                status(project=p, user=u, save=True)

        response = self.client.get(self._url(dict(since_id=10, count=20)))
        data = json.loads(response.data)
        eq_(data[19]['id'], 11)

        response = self.client.get(self._url(dict(since_id=10, count=10)))
        data = json.loads(response.data)
        eq_(data[9]['id'], 21)

        response = self.client.get(self._url(dict(since_id=10, count=30)))
        data = json.loads(response.data)
        eq_(len(data), 20)
        eq_(data[19]['id'], 11)

        response = self.client.get(self._url(dict(since_id=0)))
        eq_(response.status_code, 400)

        response = self.client.get(self._url(dict(since_id='a')))
        eq_(response.status_code, 400)

 def test_timesince(self):
     """Test the timesince last update endpoint."""
     with self.app.app_context():
         status(user_id=self.user.id, save=True)
     url = '%s?%s' % (self.url, urlencode(self.query))
     response = self.client.get(url)
     eq_(response.status_code, 200)
     eq_(response.data, '{"timesince": 0}')

    def test_status_reply_count(self):
        """Test the reply_count property of the Status model."""
        with self.app.app_context():
            u = user(save=True)
            s = status(user=u, project=None, save=True)
            for i in range(5):
                status(user=u, project=None, reply_to=s, save=True)

            eq_(s.reply_count, 5)

 def test_timeline(self):
     """Test the home_timeline endpoint"""
     with self.app.app_context():
         u = user(save=True, team={})
         p = project(save=True)
         status(user=u, project=p, save=True)
     response = self.client.get(self._url())
     eq_(response.status_code, 200)
     eq_(response.content_type, 'application/json')

    def setUp(self):
        super(HelpersTestCase, self).setUp()
        with self.app.app_context():
            u = user(username='jdoe', save=True)
            for i in range(100):
                status(project=None, user=u, save=True)

        db = get_session(self.app)
        self.query = db.query(Status).order_by(Status.id)

    def test_timeline_filters_user(self):
        """Test the timeline only shows the passed in user."""
        with self.app.app_context():
            u = user(save=True)
            status(user=u, project=None, save=True)
            u2 = user(username="janedoe", email="[email protected]", slug="janedoe", save=True)
            status(user=u2, project=None, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(len(data), 1)
        eq_(data[0]["user"], u.dictify())

    def test_paginate(self):
        """Test the paginate helper function."""
        db = get_session(self.app)

        statuses = []
        with self.app.app_context():
            p = project(save=True)
            u = user(save=True)

            # Create 100 statuses
            for i in range(30):
                statuses.append(status(project=p, user=u,
                                       created=datetime(2012, 5, 25),
                                       save=True))

            for i in range(30):
                statuses.append(status(project=p, user=u,
                                       created=datetime(2012, 6, 25),
                                       save=True))

            for i in range(40):
                statuses.append(status(project=p, user=u,
                                       created=datetime(2012, 7, 25),
                                       save=True))

        s = db.query(Status).order_by(Status.id)

        # Test simple pagination
        page = paginate(s, page=1)

        eq_(page.pages, 5)

        eq_(page.has_prev, False)
        eq_(page.has_next, True)

        page = paginate(s, page=3)
        eq_(page.has_prev, True)
        eq_(page.has_next, True)

        page = paginate(s, page=5)
        eq_(page.has_prev, True)
        eq_(page.has_next, False)

        # Test date filtered pagination
        page = paginate(s, page=1, startdate=datetime(2012, 5, 28))
        eq_(page.pages, 4)

        page = paginate(s, page=1, startdate=datetime(2012, 5, 28),
                        enddate=datetime(2012, 6, 28))
        eq_(page.pages, 2)

        page = paginate(s, page=1, enddate=datetime(2012, 6, 28))
        eq_(page.pages, 3)

    def test_status_replies(self):
        """Test the loading of replies for a status."""
        with self.app.app_context():
            p = project(save=True)
            u = user(save=True)
            s = status(project=p, user=u, save=True)

            for i in range(30):
                status(project=p, user=u, reply_to=s, save=True)

            page = s.replies()

        eq_(page.pages, 2)

    def test_timeline_trim_project(self):
        """Test the trim_project parameter of home_timeline"""
        with self.app.app_context():
            p = project(save=True)
            status(project=p, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(data[0]["project"], p.dictify())

        response = self.client.get(self._url(dict(trim_project=1)))
        data = json.loads(response.data)
        eq_(data[0]["project"], p.id)

    def test_timeline_filters_project(self):
        """Test the timeline only shows the passed in project."""
        with self.app.app_context():
            u = user(save=True)
            p = project(save=True)
            status(user=u, project=p, save=True)
            p2 = project(name="Test Project 2", slug="test-project-2", save=True)
            status(user=u, project=p2, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(len(data), 1)
        eq_(data[0]["project"], p.dictify())

    def test_timeline_trim_user(self):
        """Test the trim_user parameter of home_timeline"""
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            status(user=u, project=p, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(data[0]['user'], u.dictify())

        response = self.client.get(self._url(dict(trim_user=1)))
        data = json.loads(response.data)
        eq_(data[0]['user'], u.id)

    def test_timeline_filters_team(self):
        """Test the timeline only shows the passed in team."""
        with self.app.app_context():
            u = user(save=True, team={})
            u2 = user(username='janedoe', email='[email protected]',
                      slug='janedoe', save=True, team={'name': 'XXX',
                                                       'slug': 'xxx'})
            p = project(save=True)
            status(user=u, project=p, save=True)
            status(user=u2, project=p, save=True)

        response = self.client.get(self._url(dict(team_id=u.teams[0].id)))
        data = json.loads(response.data)
        eq_(len(data), 1)
        eq_(data[0]['user'], u.dictify())

    def test_timeline_filters_team(self):
        """Test the timeline only shows the passed in team."""
        with self.app.app_context():
            u = user(save=True, team={})
            u2 = user(
                username="janedoe", email="[email protected]", slug="janedoe", save=True, team={"name": "XXX", "slug": "xxx"}
            )
            p = project(save=True)
            status(user=u, project=p, save=True)
            status(user=u2, project=p, save=True)

        response = self.client.get(self._url(dict(team_id=u.teams[0].id)))
        data = json.loads(response.data)
        eq_(len(data), 1)
        eq_(data[0]["user"], u.dictify())

    def test_create_reply(self):
        """Test creation of replies"""
        s = status(save=True)
        sid = s.id

        data = json.dumps({
            'api_key': settings.API_KEY,
            'user': 'r1cky',
            'content': 'reply to status',
            'reply_to': sid})
        response = self.app.post(
            '/api/v1/status/', data=data, content_type='application/json')
        self.assertEqual(response.status_code, 200)

        # Verify that the status is actually a reply
        r = Status.query.filter(Status.reply_to_id==sid).first()
        self.assertEqual(r.content, 'reply to status')

        # Verify that the reply is included in the list of replies
        s = Status.query.get(sid)
        assert r in s.replies().items

        # You should not be able to reply to the reply
        data = json.dumps({
            'api_key': settings.API_KEY,
            'user': 'r1cky',
            'content': 'should not work',
            'reply_to': r.id})
        response = self.app.post(
            '/api/v1/status/', data=data, content_type='application/json')
        self.assertEqual(response.status_code, 400)

    def test_user_recent_statuses(self):
        """Test loading of recent statuses for a project."""
        with self.app.app_context():
            u = user(save=True)

            # Create 30 statuses
            for i in range(30):
                s = status(project=None, user=u, save=True)

            # Create 30 replies
            for i in range(30):
                status(project=None, user=u, reply_to=s, save=True)

            # Should not include replies
            page = u.recent_statuses()
            eq_(page.pages, 2)

    def test_status_repr(self):
        """Test the __repr__ function of the Status model."""
        with self.app.app_context():
            u = user(username='testuser', save=True)
            s = status(content='my status update', user=u, save=True)

        eq_(repr(s), '<Status: testuser: my status update>')

 def test_delete_status_missing_api_key(self):
     """Request with missing API key should return 403"""
     s = status(save=True)
     data = json.dumps({'user': s.user.username})
     response = self.app.delete(
         '/api/v1/status/%s/' % s.id, data=data,
         content_type='application/json')
     self.assertEqual(response.status_code, 403)

 def test_status_week_end(self):
     """Test the week_end function of the Status model."""
     d = datetime(2014, 5, 8, 17, 17, 51, 0)
     with self.app.app_context():
         u = user(username='testuser', save=True)
         s = status(content='my status update', created=d, user=u, save=True)
     d_actual = s.week_end.strftime("%Y-%m-%d")
     eq_(d_actual, "2014-05-11") # Happy Mother's Day!

 def test_status_weeks_at_year_start(self):
     """Test the week_{start|end} function around the start of the year."""
     d = datetime(2013, 12, 31, 12, 13, 45, 0)
     with self.app.app_context():
         u = user(username='testuser', save=True)
         s = status(content='my status update', created=d, user=u, save=True)
     eq_(s.week_start.strftime("%Y-%m-%d"), "2013-12-30")
     eq_(s.week_end.strftime("%Y-%m-%d"), "2014-01-05")