How to use the sortinghat.db.api.find_unique_identity function in sortinghat

To help you get started, we’ve selected a few sortinghat examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github chaoss / grimoirelab-sortinghat / tests / test_db_api.py View on Github external
profile = api.edit_profile(session, uidentity,
                                       name='Smith, J.', email='jsmith@example.net',
                                       is_bot=True, country_code='US',
                                       gender='male', gender_acc=98)

            self.assertIsInstance(profile, Profile)
            self.assertEqual(profile.name, 'Smith, J.')
            self.assertEqual(profile.email, 'jsmith@example.net')
            self.assertEqual(profile.is_bot, True)
            self.assertEqual(profile.country_code, 'US')
            self.assertEqual(profile.gender, 'male')
            self.assertEqual(profile.gender_acc, 98)

        with self.db.connect() as session:
            uidentity = api.find_unique_identity(session, uuid)

            profile = uidentity.profile

            self.assertEqual(profile.uuid, uuid)
            self.assertEqual(profile.name, 'Smith, J.')
            self.assertEqual(profile.email, 'jsmith@example.net')
            self.assertEqual(profile.is_bot, True)
            self.assertEqual(profile.country_code, 'US')
            self.assertIsInstance(profile.country, Country)
            self.assertEqual(profile.country.code, 'US')
            self.assertEqual(profile.country.name, 'United States of America')
            self.assertEqual(profile.gender, 'male')
            self.assertEqual(profile.gender_acc, 98)
github chaoss / grimoirelab-sortinghat / tests / test_db_api.py View on Github external
def test_from_date_none(self):
        """Check if an enrollment cannot be withdrawn when from_date is None"""

        with self.db.connect() as session:
            uidentity = api.add_unique_identity(session, '1234567890ABCDFE')
            org = api.add_organization(session, 'Example')
            api.enroll(session, uidentity, org,
                       from_date=datetime.datetime(1999, 1, 1),
                       to_date=datetime.datetime(2000, 1, 1))

        with self.db.connect() as session:
            uidentity = api.find_unique_identity(session, '1234567890ABCDFE')
            org = api.find_organization(session, 'Example')

            with self.assertRaisesRegex(ValueError, FROM_DATE_NONE_ERROR):
                api.withdraw(session, uidentity, org,
                             from_date=None,
                             to_date=datetime.datetime(1999, 1, 1))
github chaoss / grimoirelab-sortinghat / tests / test_db_api.py View on Github external
session.add(enrol)
            enrol = Enrollment(uidentity=jdoe, organization=example_org)
            session.add(enrol)
            enrol = Enrollment(uidentity=jdoe, organization=bitergia_org)
            session.add(enrol)

        before_dt = datetime.datetime.utcnow()

        with self.db.connect() as session:
            organization = api.find_organization(session, 'Example')
            api.delete_organization(session, organization)

        after_dt = datetime.datetime.utcnow()

        with self.db.connect() as session:
            jsmith = api.find_unique_identity(session, 'AAAA')
            self.assertIsInstance(jsmith.last_modified, datetime.datetime)
            self.assertLessEqual(before_dt, jsmith.last_modified)
            self.assertGreaterEqual(after_dt, jsmith.last_modified)

            jdoe = api.find_unique_identity(session, 'BBBB')
            self.assertIsInstance(jdoe.last_modified, datetime.datetime)
            self.assertLessEqual(before_dt, jdoe.last_modified)
            self.assertGreaterEqual(after_dt, jdoe.last_modified)
github chaoss / grimoirelab-sortinghat / tests / test_db_api.py View on Github external
def test_unique_identity_not_found(self):
        """Test whether the function returns None when the unique identity is not found"""

        with self.db.connect() as session:
            uuid = 'abcdefghijklmnopqrstuvwxyz'
            uidentity = UniqueIdentity(uuid=uuid)
            session.add(uidentity)

        with self.db.connect() as session:
            uidentity = api.find_unique_identity(session, 'zyxwuv')
            self.assertEqual(uidentity, None)
github chaoss / grimoirelab-sortinghat / tests / test_db_api.py View on Github external
before_dt = datetime.datetime.utcnow()

        with self.db.connect() as session:
            uidentity = api.find_unique_identity(session, 'BBBB')

            enrollment = session.query(Enrollment).\
                filter(Enrollment.uuid == 'AAAA').first()

            api.move_enrollment(session, enrollment, uidentity)

        after_dt = datetime.datetime.utcnow()

        with self.db.connect() as session:
            uidentity1 = api.find_unique_identity(session, 'AAAA')
            uidentity2 = api.find_unique_identity(session, 'BBBB')

            self.assertIsInstance(uidentity1.last_modified, datetime.datetime)
            self.assertLessEqual(before_dt, uidentity1.last_modified)
            self.assertGreaterEqual(after_dt, uidentity1.last_modified)

            self.assertIsInstance(uidentity2.last_modified, datetime.datetime)
            self.assertLessEqual(before_dt, uidentity2.last_modified)
            self.assertGreaterEqual(after_dt, uidentity2.last_modified)
github chaoss / grimoirelab-sortinghat / tests / test_db_api.py View on Github external
def test_last_modified(self):
        """Check if last modification date is updated"""

        uuid = '1234567890ABCDFE'

        before_dt = datetime.datetime.utcnow()

        with self.db.connect() as session:
            api.add_unique_identity(session, uuid)

        after_dt = datetime.datetime.utcnow()

        with self.db.connect() as session:
            uidentity = api.find_unique_identity(session, uuid)
            self.assertIsInstance(uidentity.last_modified, datetime.datetime)
            self.assertLessEqual(before_dt, uidentity.last_modified)
            self.assertGreaterEqual(after_dt, uidentity.last_modified)
github chaoss / grimoirelab-sortinghat / tests / test_db_api.py View on Github external
def test_last_modified(self):
        """Check if last modification date is updated"""

        with self.db.connect() as session:
            uidentity = api.add_unique_identity(session, 'AAAA')
            api.add_unique_identity(session, 'BBBB')

            org = api.add_organization(session, 'Example')
            api.enroll(session, uidentity, org)

        before_dt = datetime.datetime.utcnow()

        with self.db.connect() as session:
            uidentity = api.find_unique_identity(session, 'BBBB')

            enrollment = session.query(Enrollment).\
                filter(Enrollment.uuid == 'AAAA').first()

            api.move_enrollment(session, enrollment, uidentity)

        after_dt = datetime.datetime.utcnow()

        with self.db.connect() as session:
            uidentity1 = api.find_unique_identity(session, 'AAAA')
            uidentity2 = api.find_unique_identity(session, 'BBBB')

            self.assertIsInstance(uidentity1.last_modified, datetime.datetime)
            self.assertLessEqual(before_dt, uidentity1.last_modified)
            self.assertGreaterEqual(after_dt, uidentity1.last_modified)
github chaoss / grimoirelab-sortinghat / tests / test_db_api.py View on Github external
with self.db.connect() as session:
            # Check everything is in place and remove
            jsmith = api.find_unique_identity(session, 'AAAA')
            self.assertIsInstance(jsmith.profile, Profile)
            self.assertEqual(len(jsmith.identities), 2)

            enrollments = session.query(Enrollment).\
                join(UniqueIdentity, Organization).\
                filter(UniqueIdentity.uuid == 'AAAA').\
                order_by(Enrollment.start).all()
            self.assertEqual(len(enrollments), 2)

            api.delete_unique_identity(session, jsmith)

        with self.db.connect() as session:
            jsmith = api.find_unique_identity(session, 'AAAA')
            self.assertEqual(jsmith, None)

            jsmith_id = api.find_identity(session, '0001')
            self.assertEqual(jsmith_id, None)

            jsmith_id = api.find_identity(session, '0002')
            self.assertEqual(jsmith_id, None)

            profile = session.query(Profile).\
                join(UniqueIdentity).\
                filter(UniqueIdentity.uuid == 'AAAA').\
                first()
            self.assertEqual(profile, None)

            enrollments = session.query(Enrollment).\
                join(UniqueIdentity, Organization).\
github chaoss / grimoirelab-sortinghat / sortinghat / api.py View on Github external
if uuid is None:
        raise InvalidValueError('uuid cannot be None')
    if uuid == '':
        raise InvalidValueError('uuid cannot be an empty string')
    if organization is None:
        raise InvalidValueError('organization cannot be None')
    if organization == '':
        raise InvalidValueError('organization cannot be an empty string')

    if not from_date:
        from_date = MIN_PERIOD_DATE
    if not to_date:
        to_date = MAX_PERIOD_DATE

    with db.connect() as session:
        uidentity = find_unique_identity(session, uuid)

        if not uidentity:
            raise NotFoundError(entity=uuid)

        org = find_organization(session, organization)

        if not org:
            raise NotFoundError(entity=organization)

        try:
            enroll_db(session, uidentity, org,
                      from_date=from_date, to_date=to_date)
        except ValueError as e:
            raise InvalidValueError(e)
github chaoss / grimoirelab-sortinghat / sortinghat / api.py View on Github external
When 'from_uuid' and 'to_uuid' are equal, the action does not have any
    effect.

    The function raises a 'NotFoundError exception when either 'from_uuid'
    or 'to_uuid' do not exist in the registry.

    :param from_uuid: identifier of the unique identity set to merge
    :param to_uuid: identifier of the unique identity where 'from_uuid'
        will be merged

    :raises NotFoundError: raised when either 'from_uuid' or 'to_uuid'
        do not exist in the registry
    """
    with db.connect() as session:
        fuid = find_unique_identity(session, from_uuid)
        tuid = find_unique_identity(session, to_uuid)

        if not fuid:
            raise NotFoundError(entity=from_uuid)

        if from_uuid == to_uuid:
            return

        if not tuid:
            raise NotFoundError(entity=to_uuid)

        # Update profile information
        if tuid.profile and fuid.profile:
            # Update data giving priority to 'to_uuid'.
            # When 'is_bot' is set to True in any of the unique identities
            # it will remain the same.