notes.txt

The command to run tests: `python3 -m unittest stream_tests.py`

Notes:
1. It took ~1 hour for task implementing and writing notes.
2. The project was completed with TDD technology, tests first.
3. I didn't implement any limiters, like input string size or stream maximum size limiters. I assumed that these limiters were outside of the task scope.
4. I spent some time deciding what better to use for string parsing, regex, or simple cycle iteration. In the end, the cycle was preferred because of better readability.
5. It wasn't clear from requirements what exactly filter by date should be implemented, so I implemented `get_latest_events_created_after_date()` method.

stream.py

import datetime


class EventStream:
    _events = None

    def __init__(self):
        self._events = []

    def push_event(self, event_text):
        """Add a new event to the top of the _events stream"""
        if not event_text or type(event_text) != str:
            return False

        parsed_event = self._parse_event_text(event_text)
        # put a new event on the top
        self._events.insert(0, parsed_event)

        return True

    def get_latest_events_by_hashtag(self, hashtag, limit=10):
        found_events = [x for x in self._events if hashtag in x['hashtags']]
        return found_events[:limit]

    def get_latest_events_by_mention(self, mention, limit=10):
        found_events = [x for x in self._events if mention in x['mentions']]
        return found_events[:limit]

    def get_latest_events_created_after_date(self, date, limit=10):
        found_events = [x for x in self._events if x['utc_date'] > date]
        return found_events[:limit]

    def get_latest_events(self, limit=10):
        return self._events[:limit]

    def _parse_event_text(self, event_text):
        hashtags, mentions = [], []
        for word in event_text.split():
            if word.startswith('#'):
                hashtags.append(word)

            if word.startswith('@'):
                mentions.append(word)

        return {
            'text': event_text,
            'hashtags': hashtags,
            'mentions': mentions,
            'utc_date': datetime.datetime.utcnow()
        }

stream_tests.py

from datetime import datetime, timedelta
from unittest import TestCase

from stream import EventStream


class EventStreamUnitTest(TestCase):

    def setUp(self):
        self.stream = EventStream()

    def test_push_empty_event(self):
        event_text = ''
        result = self.stream.push_event(event_text)
        self.assertEqual(result, False)

    def test_parse_hashtags(self):
        event_text = 'some #first1 text with #second hashtags'
        result = self.stream.push_event(event_text)
        self.assertEqual(result, True)

        event = self.stream.get_latest_events()[0]
        self.assertEqual(event['hashtags'], ['#first1', '#second'])

    def test_parse_mentions(self):
        event_text = 'some @all text with @all-friends'
        result = self.stream.push_event(event_text)
        self.assertEqual(result, True)

        event = self.stream.get_latest_events()[0]
        self.assertEqual(event['mentions'], ['@all', '@all-friends'])

    def test_get_events_by_hashtag(self):
        self.stream.push_event('one #first')
        self.stream.push_event('two #second')
        self.stream.push_event('three #first')

        events = self.stream.get_latest_events_by_hashtag('#first')
        self.assertEqual(len(events), 2)
        self.assertEqual(events[0]['text'], 'three #first')
        self.assertEqual(events[1]['text'], 'one #first')

    def test_get_events_by_mention(self):
        self.stream.push_event('one @all')
        self.stream.push_event('two @john')
        self.stream.push_event('three @john')

        events = self.stream.get_latest_events_by_mention('@john')
        self.assertEqual(len(events), 2)
        self.assertEqual(events[0]['text'], 'three @john')
        self.assertEqual(events[1]['text'], 'two @john')

    def test_get_events_by_date(self):
        self.stream.push_event('one @all')
        self.stream.push_event('two @john')
        self.stream.push_event('three @john')

        date = datetime.utcnow() - timedelta(days=1)
        events = self.stream.get_latest_events_created_after_date(date)
        self.assertEqual(len(events), 3)

        date = datetime(2100, 1, 1)
        events = self.stream.get_latest_events_created_after_date(date)
        self.assertEqual(len(events), 0)