def test_cancelled_event(db, default_account):
    with open(absolute_path(FIXTURES + 'google_cancelled1.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, ics_data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").one()

    assert ev.status == 'confirmed'

    with open(absolute_path(FIXTURES + 'google_cancelled2.ics')) as fd:
        ics_data = fd.read()

    msg2 = add_fake_msg_with_calendar_part(
        db.session, default_account, ics_data)

    import_attached_events(db.session, default_account, msg2)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").one()

    assert ev.status == 'cancelled'

def test_icloud_cancelled_event(db, default_account):
    with open(absolute_path(FIXTURES + 'icloud_cancelled1.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, ics_data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "5919D444-7C99-4687-A526-FC5D10091318").one()

    assert ev.status == 'confirmed'

    with open(absolute_path(FIXTURES + 'icloud_cancelled2.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, ics_data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "5919D444-7C99-4687-A526-FC5D10091318").one()

    assert ev.status == 'cancelled'

def test_event_update(db, default_account, message):
    add_fake_calendar(db.session, default_account.namespace.id,
                      name="Emailed events", read_only=True)

    with open(absolute_path(FIXTURES + 'gcal_v1.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, ics_data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").one()

    assert ev.location == ("Olympia Hall, 28 Boulevard des Capucines, "
                           "75009 Paris, France")

    with open(absolute_path(FIXTURES + 'gcal_v2.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, ics_data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").one()

    assert ev.location == (u"Le Zenith, 211 Avenue Jean Jaures, "
                            "75019 Paris, France")

def test_self_sent_update(db, default_account, message):

    # Create the calendars
    add_fake_calendar(db.session, default_account.namespace.id,
                      name="Emailed events", read_only=True)

    default_calendar = add_fake_calendar(db.session,
                                         default_account.namespace.id,
                                         name="Calendar", read_only=False)

    # Import the self-sent event.
    with open(absolute_path(FIXTURES + 'self_sent_v1.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(db.session, default_account,
                                          ics_data)
    msg.from_addr = [(default_account.name, default_account.email_address)]
    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    evs = db.session.query(Event).filter(
        Event.uid == "[email protected]").all()

    assert len(evs) == 1
    ev = evs[0]
    assert ev.location == ("Olympia Hall, 28 Boulevard des Capucines, "
                           "75009 Paris, France")

    # Create a copy of the event, and store it in the default calendar.
    event_copy = Event()
    event_copy.update(ev)
    event_copy.calendar = default_calendar
    db.session.add(event_copy)
    db.session.commit()

    with open(absolute_path(FIXTURES + 'self_sent_v2.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, ics_data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    evs = db.session.query(Event).filter(
        Event.uid == "[email protected]").all()

    # Check that the event in the default calendar didn't get updated.
    assert len(evs) == 2
    for ev in evs:
        db.session.refresh(ev)
        if ev.calendar_id == default_calendar.id:
            assert ev.location == ("Olympia Hall, 28 Boulevard des Capucines, "
                                   "75009 Paris, France")
        else:
            assert ev.location == (u"Le Zenith, 211 Avenue Jean Jaures, "
                                   "75019 Paris, France")

def test_multiple_summaries(db, default_account):
    data = None
    with open(absolute_path(FIXTURES + 'multiple_summaries.ics')) as fd:
        data = fd.read()

    events = events_from_ics(default_account.namespace,
                             default_account.emailed_events_calendar, data)

    assert len(events) == 1
    assert events[0].title == 'The Strokes - Is this it?'

def test_event_no_participants(db, default_account):
    data = None
    with open(absolute_path(FIXTURES + 'event_with_no_participants.ics')) as fd:
        data = fd.read()

    events = events_from_ics(default_account.namespace,
                             default_account.emailed_events_calendar, data)
    assert len(events) == 1, "There should be only one event in the test file"
    ev = events[0]
    assert len(ev.participants) == 0

def test_handle_missing_sequence_number(db, default_account):
    with open(absolute_path(FIXTURES + 'event_without_sequence.ics')) as fd:
        data = fd.read()

    events = events_from_ics(default_account.namespace,
                             default_account.emailed_events_calendar, data)
    events = events['invites']
    assert len(events) == 1
    ev = events[0]
    assert ev.sequence_number == 0

def test_event_no_end_time(db, default_account):
    # With no end time, import should fail
    with open(absolute_path(FIXTURES + 'meetup_infinite.ics')) as fd:
        ics_data = fd.read()

    add_fake_msg_with_calendar_part(db.session, default_account, ics_data)

    # doesn't raise an exception (to not derail message parsing, but also
    # doesn't create an event)
    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").first()
    assert not ev

def test_iphone_through_exchange(db, default_account):
    data = None
    with open(absolute_path(FIXTURES + 'iphone_through_exchange.ics')) as fd:
        data = fd.read()

    events = events_from_ics(default_account.namespace,
                             default_account.emailed_events_calendar, data)
    assert len(events) == 1, "There should be only one event in the test file"

    ev = events[0]
    assert ev.start == arrow.get(2014, 12, 27, 15, 0)
    assert ev.end == arrow.get(2014, 12, 27, 16, 0)

def test_multiple_events(db, default_account):
    data = None
    with open(absolute_path(FIXTURES + 'multiple_events.ics')) as fd:
        data = fd.read()

    events = events_from_ics(default_account.namespace,
                             default_account.emailed_events_calendar, data)
    assert len(events) == 2
    ev0 = events[0]
    ev1 = events[1]
    assert len(ev0.participants) == 0
    assert len(ev1.participants) == 0

    assert ev1.start == arrow.get(2015, 03, 17, 0, 0)

def test_windows_tz_ical(db, default_account):
    data = None
    with open(absolute_path(FIXTURES + 'windows_event.ics')) as fd:
        data = fd.read()

    events = events_from_ics(default_account.namespace,
                             default_account.emailed_events_calendar, data)
    assert len(events) == 1, "There should be only one event in the test file"

    ev = events[0]
    assert ev.start == arrow.get(2015, 2, 20, 8, 30)
    assert ev.end == arrow.get(2015, 2, 20, 9, 0)
    assert ev.title == "Pommes"
    assert len(ev.participants) == 1
    assert ev.participants[0]['email'] == '[email protected]'

def test_icloud_allday_event(db, default_account):
    data = None
    with open(absolute_path(FIXTURES + 'icloud_oneday_event.ics')) as fd:
        data = fd.read()

    events = events_from_ics(default_account.namespace,
                             default_account.emailed_events_calendar, data)
    assert len(events) == 1, "There should be only one event in the test file"

    ev = events[0]
    assert ev.all_day is True
    assert ev.start == arrow.get(2015, 3, 16, 0, 0)
    assert ev.end == arrow.get(2015, 3, 17, 0, 0)

    assert len(ev.participants) == 2
    assert ev.participants[0]['email'] == '[email protected]'

def test_recurring_ical(db, default_account):
    with open(absolute_path(FIXTURES + 'gcal_recur.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, ics_data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").one()

    assert isinstance(ev, RecurringEvent)
    assert isinstance(ev.recurring, list)
    assert ev.start_timezone == 'America/Los_Angeles'

def test_invalid_rsvp(db, default_account):
    # Test that we don't save an RSVP reply with an invalid id.
    data = None
    with open(absolute_path(FIXTURES + 'invalid_rsvp.ics')) as fd:
        data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").all()

    assert len(ev) == 0

def test_truncate_bogus_sequence_numbers(db, default_account):
    data = None
    with open(absolute_path(FIXTURES + 'bogus_sequence_number.ics')) as fd:
        data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").one()

    # Check that the sequence number got truncated to the biggest possible
    # number.
    assert ev.sequence_number == 2147483647L

def test_rsvp_for_other_provider(db, default_account):
    # Test that we don't save RSVP replies which aren't replies to a Nylas
    # invite.
    data = None
    with open(absolute_path(FIXTURES + 'invalid_rsvp2.ics')) as fd:
        data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, default_account, data)

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    ev = db.session.query(Event).filter(
        Event.uid == "[email protected]").all()

    assert len(ev) == 0

def test_invalid_sender(from_value, db, default_account):
    # Check that messages with an invalid from field get discarded.
    data = None
    event_count = db.session.query(Event).count()

    with open(absolute_path(FIXTURES + 'iphone_through_exchange.ics')) as fd:
        data = fd.read()

    msg = add_fake_msg_with_calendar_part(db.session, default_account,
                                          data)
    msg.from_addr = from_value
    db.session.add(msg)
    db.session.commit()

    import_attached_events(db.session, default_account, msg)
    db.session.commit()

    event_count2 = db.session.query(Event).count()
    assert event_count == event_count2, "The event shouldn't have been added"

def test_icalendar_import(db, generic_account, message):
    add_fake_calendar(db.session, generic_account.namespace.id,
                      name="Emailed events", read_only=True)

    with open(absolute_path(FIXTURES + 'invite_w_rsvps1.ics')) as fd:
        ics_data = fd.read()

    msg = add_fake_msg_with_calendar_part(
        db.session, generic_account, ics_data)

    import_attached_events(db.session, generic_account, msg)

    ev = db.session.query(Event).filter(
        Event.uid == ("040000008200E00074C5B7101A82E00800000000"
                      "F9125A30B06BD001000000000000000010000000"
                      "9D791C7548BFD144BFA54F14213CAD25")).one()

    assert len(ev.participants) == 2
    for participant in ev.participants:
        assert participant['status'] == 'noreply'

def pytest_generate_tests(metafunc):
    if 'db' in metafunc.fixturenames:
        dumpfile = absolute_path(config()['BASE_DUMP'])
        savedb = False

        metafunc.parametrize('db', [(dumpfile, savedb)], indirect=True)

def latest_db(config):
    dumpfile = absolute_path('data/base_dump.sql')
    testdb = TestDB(config, dumpfile)
    yield testdb
    testdb.teardown()