@Test public void testAssignUsersInBatchCalls() { Experiment experiment = Experiment.withID(Experiment.ID.valueOf(this.experimentId)) .withIsPersonalizationEnabled(false) .withIsRapidExperiment(false).build(); User.ID userID1 = User.ID.valueOf("testuser1"); User.ID userID2 = User.ID.valueOf("testuser2"); Context context = Context.valueOf("test"); Date date = new Date(); String bucketLabel = "bucket-1"; Assignment assignment1 = Assignment.newInstance(experiment.getID()) .withBucketLabel(Bucket.Label.valueOf(bucketLabel)) .withCreated(date) .withApplicationName(APPLICATION_NAME) .withContext(context) .withUserID(userID1) .build(); Assignment assignment2 = Assignment.newInstance(experiment.getID()) .withBucketLabel(null) .withCreated(date) .withApplicationName(APPLICATION_NAME) .withContext(context) .withUserID(userID2) .build(); List<Pair<Experiment, Assignment>> assignmentPairs = new LinkedList<>(); assignmentPairs.add(new ImmutablePair<>(experiment, assignment1)); assignmentPairs.add(new ImmutablePair<>(experiment, assignment2)); ResultSetFuture genericResultSetFuture = mock(ResultSetFuture.class); ResultSet genericResultSet = mock(ResultSet.class); when(genericResultSetFuture.getUninterruptibly()).thenReturn(genericResultSet); doNothing().when(assignmentsCountExecutor).execute(any()); when(driver.getSession()).thenReturn(mock(Session.class)); when(driver.getSession().execute(any(BatchStatement.class))).thenReturn(genericResultSet); boolean success = true; try { repository.assignUsersInBatch(assignmentPairs, date); } catch (Exception e) { logger.error("Failed to execute assignUser test...", e); success = false; } assertThat(success, is(true)); }
@Override @Timed public void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date) { incrementCounts(assignments, date); indexExperimentsToUser(assignments); }
CassandraAssignmentsRepository implements AssignmentsRepository { @Override @Timed public void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date) { incrementCounts(assignments, date); indexExperimentsToUser(assignments); } }
CassandraAssignmentsRepository implements AssignmentsRepository { @Override @Timed public void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date) { incrementCounts(assignments, date); indexExperimentsToUser(assignments); } @Inject  CassandraAssignmentsRepository(
            @CassandraRepository ExperimentRepository experimentRepository,
            @DatabaseRepository ExperimentRepository dbRepository,
            EventLog eventLog,
            ExperimentAccessor experimentAccessor,
            ExperimentUserIndexAccessor experimentUserIndexAccessor,

            UserAssignmentExportAccessor userAssignmentExportAccessor,

            BucketAccessor bucketAccessor,
            BucketAssignmentCountAccessor bucketAssignmentCountAccessor,

            StagingAccessor stagingAccessor,
            PrioritiesAccessor prioritiesAccessor,
            ExclusionAccessor exclusionAccessor,
            PageExperimentIndexAccessor pageExperimentIndexAccessor,
            CassandraDriver driver,
            MappingManager mappingManager,
            @Named("AssignmentsCountThreadPoolExecutor") ThreadPoolExecutor assignmentsCountExecutor,
            final @Named("assign.user.to.export") boolean assignUserToExport,
            final @Named("assign.bucket.count") boolean assignBucketCount,
            final @Named("default.time.format") String defaultTimeFormat); }
CassandraAssignmentsRepository implements AssignmentsRepository { @Override @Timed public void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date) { incrementCounts(assignments, date); indexExperimentsToUser(assignments); } @Inject  CassandraAssignmentsRepository(
            @CassandraRepository ExperimentRepository experimentRepository,
            @DatabaseRepository ExperimentRepository dbRepository,
            EventLog eventLog,
            ExperimentAccessor experimentAccessor,
            ExperimentUserIndexAccessor experimentUserIndexAccessor,

            UserAssignmentExportAccessor userAssignmentExportAccessor,

            BucketAccessor bucketAccessor,
            BucketAssignmentCountAccessor bucketAssignmentCountAccessor,

            StagingAccessor stagingAccessor,
            PrioritiesAccessor prioritiesAccessor,
            ExclusionAccessor exclusionAccessor,
            PageExperimentIndexAccessor pageExperimentIndexAccessor,
            CassandraDriver driver,
            MappingManager mappingManager,
            @Named("AssignmentsCountThreadPoolExecutor") ThreadPoolExecutor assignmentsCountExecutor,
            final @Named("assign.user.to.export") boolean assignUserToExport,
            final @Named("assign.bucket.count") boolean assignBucketCount,
            final @Named("default.time.format") String defaultTimeFormat); @Override @Timed void populateAssignmentsMetadata(User.ID userID, Application.Name appName, Context context, ExperimentBatch experimentBatch, Optional<Map<Experiment.ID, Boolean>> allowAssignments,
                                            PrioritizedExperimentList prioritizedExperimentList,
                                            Map<Experiment.ID, com.intuit.wasabi.experimentobjects.Experiment> experimentMap,
                                            Map<Experiment.ID, BucketList> bucketMap,
                                            Map<Experiment.ID, List<Experiment.ID>> exclusionMap
    ); @Override @Timed List<Pair<Experiment, String>> getAssignments(User.ID userID,
                                                         Application.Name appLabel,
                                                         Context context,
                                                         Map<Experiment.ID, Experiment> experimentMap); @Override @Timed Assignment getAssignment(User.ID userID, Application.Name appName, Experiment.ID experimentID, Context context); @Override @Timed void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date); @Override void assignUserToExports(Assignment assignment, Date date); @Override void deleteAssignment(Experiment experiment, User.ID userID, Context context, Application.Name appName, Assignment currentAssignment); @Override StreamingOutput getAssignmentStream(Experiment.ID experimentID, Context context, Parameters parameters, Boolean ignoreNullBucket); @Override @Timed void pushAssignmentToStaging(String type, String exception, String data); void pushAssignmentsToStaging(String type, String exception, Collection<String> data); Map<UUID, String> getBatchPayloadsFromStaging(int batchSize); void deleteFromStaging(UUID timeUUID); @Override void updateBucketAssignmentCount(Experiment experiment, Assignment assignment, boolean countUp); @Override AssignmentCounts getBucketAssignmentCount(Experiment experiment); @Override Map<Experiment.ID, AssignmentCounts> getBucketAssignmentCountsInParallel(List<Experiment.ID> experimentIds); }
CassandraAssignmentsRepository implements AssignmentsRepository { @Override @Timed public void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date) { incrementCounts(assignments, date); indexExperimentsToUser(assignments); } @Inject  CassandraAssignmentsRepository(
            @CassandraRepository ExperimentRepository experimentRepository,
            @DatabaseRepository ExperimentRepository dbRepository,
            EventLog eventLog,
            ExperimentAccessor experimentAccessor,
            ExperimentUserIndexAccessor experimentUserIndexAccessor,

            UserAssignmentExportAccessor userAssignmentExportAccessor,

            BucketAccessor bucketAccessor,
            BucketAssignmentCountAccessor bucketAssignmentCountAccessor,

            StagingAccessor stagingAccessor,
            PrioritiesAccessor prioritiesAccessor,
            ExclusionAccessor exclusionAccessor,
            PageExperimentIndexAccessor pageExperimentIndexAccessor,
            CassandraDriver driver,
            MappingManager mappingManager,
            @Named("AssignmentsCountThreadPoolExecutor") ThreadPoolExecutor assignmentsCountExecutor,
            final @Named("assign.user.to.export") boolean assignUserToExport,
            final @Named("assign.bucket.count") boolean assignBucketCount,
            final @Named("default.time.format") String defaultTimeFormat); @Override @Timed void populateAssignmentsMetadata(User.ID userID, Application.Name appName, Context context, ExperimentBatch experimentBatch, Optional<Map<Experiment.ID, Boolean>> allowAssignments,
                                            PrioritizedExperimentList prioritizedExperimentList,
                                            Map<Experiment.ID, com.intuit.wasabi.experimentobjects.Experiment> experimentMap,
                                            Map<Experiment.ID, BucketList> bucketMap,
                                            Map<Experiment.ID, List<Experiment.ID>> exclusionMap
    ); @Override @Timed List<Pair<Experiment, String>> getAssignments(User.ID userID,
                                                         Application.Name appLabel,
                                                         Context context,
                                                         Map<Experiment.ID, Experiment> experimentMap); @Override @Timed Assignment getAssignment(User.ID userID, Application.Name appName, Experiment.ID experimentID, Context context); @Override @Timed void assignUsersInBatch(List<Pair<Experiment, Assignment>> assignments, Date date); @Override void assignUserToExports(Assignment assignment, Date date); @Override void deleteAssignment(Experiment experiment, User.ID userID, Context context, Application.Name appName, Assignment currentAssignment); @Override StreamingOutput getAssignmentStream(Experiment.ID experimentID, Context context, Parameters parameters, Boolean ignoreNullBucket); @Override @Timed void pushAssignmentToStaging(String type, String exception, String data); void pushAssignmentsToStaging(String type, String exception, Collection<String> data); Map<UUID, String> getBatchPayloadsFromStaging(int batchSize); void deleteFromStaging(UUID timeUUID); @Override void updateBucketAssignmentCount(Experiment experiment, Assignment assignment, boolean countUp); @Override AssignmentCounts getBucketAssignmentCount(Experiment experiment); @Override Map<Experiment.ID, AssignmentCounts> getBucketAssignmentCountsInParallel(List<Experiment.ID> experimentIds); static final Bucket.Label NULL_LABEL; }