paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 35 Packages Projects Releases Wiki Activity

catalog-backend: more tests for DB + fix ref handling during conflicts

Browse Source 
Co-authored-by: blam <ben@blam.sh>
Co-authored-by: Johan Haals <johan.haals@gmail.com>
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
This commit is contained in:
Patrik Oldsberg
2021-06-29 14:07:19 +02:00
parent 72849d6495
commit 057e78ebb1
2 changed files with 107 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+94 -5
View File
@@ -19,6 +19,7 @@ import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
import { JsonObject } from '@backstage/config';
import { Knex } from 'knex';
import { Logger } from 'winston';
import * as uuid from 'uuid';
import { DatabaseManager } from './DatabaseManager';
import { DefaultProcessingDatabase } from './DefaultProcessingDatabase';
@@ -29,12 +30,15 @@ import {
} from './tables';
describe('Default Processing Database', () => {
const logger = getVoidLogger();
const defaultLogger = getVoidLogger();
const databases = TestDatabases.create({
ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
});
async function createDatabase(databaseId: TestDatabaseId) {
async function createDatabase(
databaseId: TestDatabaseId,
logger: Logger = defaultLogger,
) {
const knex = await databases.init(databaseId);
await DatabaseManager.createDatabase(knex);
return {
@@ -74,7 +78,11 @@ describe('Default Processing Database', () => {
it.each(databases.eachSupportedId())(
'updates refresh state with varying location keys, %p',
async databaseId => {
const { db } = await createDatabase(databaseId);
const mockWarn = jest.fn();
const { db } = await createDatabase(databaseId, ({
debug: jest.fn(),
warn: mockWarn,
} as unknown) as Logger);
await db.transaction(async tx => {
const knexTx = tx as Knex.Transaction;
@@ -102,12 +110,14 @@ describe('Default Processing Database', () => {
expectedLocationKey: 'x',
type: 'd',
expectedType: 'c',
expectConflict: true,
},
{
locationKey: undefined,
expectedLocationKey: 'x',
type: 'e',
expectedType: 'c',
expectConflict: true,
},
{
locationKey: 'x',
@@ -117,6 +127,8 @@ describe('Default Processing Database', () => {
},
];
for (const step of steps) {
mockWarn.mockClear();
await db.addUnprocessedEntities(tx, {
sourceKey: 'testing',
entities: [
@@ -127,6 +139,16 @@ describe('Default Processing Database', () => {
],
});
if (step.expectConflict) {
// eslint-disable-next-line jest/no-conditional-expect
expect(mockWarn).toHaveBeenCalledWith(
expect.stringMatching(/^Detected conflicting entityRef/),
);
} else {
// eslint-disable-next-line jest/no-conditional-expect
expect(mockWarn).not.toHaveBeenCalled();
}
const entities = await knexTx<DbRefreshStateRow>(
'refresh_state',
).select();
@@ -447,7 +469,7 @@ describe('Default Processing Database', () => {
},
kind: 'Location',
} as Entity,
locationKey: 'file:/tmp/foobar',
locationKey: 'file:///tmp/foobar',
},
],
}),
@@ -666,7 +688,7 @@ describe('Default Processing Database', () => {
},
kind: 'Location',
} as Entity,
locationKey: 'file:/tmp/foobar',
locationKey: 'file:///tmp/foobar',
},
],
});
@@ -803,6 +825,73 @@ describe('Default Processing Database', () => {
},
60_000,
);
it.each(databases.eachSupportedId())(
'should update the location key during full replace, %p',
async databaseId => {
const { knex, db } = await createDatabase(databaseId);
await createLocations(knex, ['location:default/removed']);
await insertRefreshStateRow(knex, {
entity_id: uuid.v4(),
entity_ref: 'location:default/replaced',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: '2021-04-01 13:37:00',
last_discovery_at: '2021-04-01 13:37:00',
location_key: 'file:///tmp/old',
});
await insertRefRow(knex, {
source_key: 'lols',
target_entity_ref: 'location:default/removed',
});
await insertRefRow(knex, {
source_key: 'lols',
target_entity_ref: 'location:default/replaced',
});
await db.transaction(async tx => {
await db.replaceUnprocessedEntities(tx, {
type: 'full',
sourceKey: 'lols',
items: [
{
entity: {
apiVersion: '1.0.0',
metadata: {
name: 'replaced',
},
kind: 'Location',
} as Entity,
locationKey: 'file:///tmp/foobar',
},
],
});
});
const currentRefreshState = await knex<DbRefreshStateRow>(
'refresh_state',
).select();
expect(currentRefreshState).toEqual([
expect.objectContaining({
entity_ref: 'location:default/replaced',
location_key: 'file:///tmp/foobar',
}),
]);
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
expect(currentRefRowState).toEqual([
expect.objectContaining({
source_key: 'lols',
target_entity_ref: 'location:default/replaced',
}),
]);
},
60_000,
);
});
describe('getProcessableEntities', () => {
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+13 -12
View File
@@ -325,6 +325,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
// Keeps track of the entities that we end up inserting to update refresh_state_references afterwards
const stateReferences = new Array<string>();
const conflictingStateReferences = new Array<string>();
// Upsert all of the unprocessed entities into the refresh_state table, by
// their entity ref.
@@ -352,8 +353,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
if (refreshResult === 0) {
// In the event that we can't update an existing refresh state, we first try to insert a new row
const result = await tx<DbRefreshStateRow>('refresh_state')
.insert({
try {
await tx<DbRefreshStateRow>('refresh_state').insert({
entity_id: uuid(),
entity_ref: entityRef,
unprocessed_entity: serializedEntity,
@@ -361,14 +362,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
location_key: locationKey,
next_update_at: tx.fn.now(),
last_discovery_at: tx.fn.now(),
})
.onConflict('entity_ref')
.ignore();
// If the row can't be inserted, we have a conflict, but it could be either
// because of a conflicting locationKey or a race with another instance, so check for that too
if (result.length === 0) {
// Check for conflicting entity with same entityRef but different locationKey
});
} catch (error) {
// If the row can't be inserted, we have a conflict, but it could be either
// because of a conflicting locationKey or a race with another instance, so check
// whether the conflicting entity has the same entityRef but a different locationKey
const [conflictingEntity] = await tx<DbRefreshStateRow>(
'refresh_state',
)
@@ -380,6 +378,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
this.options.logger.warn(
`Detected conflicting entityRef ${entityRef} already referenced by ${conflictingEntity.location_key} and now also ${locationKey}`,
);
conflictingStateReferences.push(entityRef);
continue;
}
}
@@ -392,7 +391,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
// Replace all references for the originating entity or source and then create new ones
if ('sourceKey' in options) {
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
.where({ source_key: options.sourceKey })
.whereNotIn('target_entity_ref', conflictingStateReferences)
.andWhere({ source_key: options.sourceKey })
.delete();
await tx.batchInsert(
'refresh_state_references',
@@ -404,7 +404,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
);
} else {
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
.where({ source_entity_ref: options.sourceEntityRef })
.whereNotIn('target_entity_ref', conflictingStateReferences)
.andWhere({ source_entity_ref: options.sourceEntityRef })
.delete();
await tx.batchInsert(
'refresh_state_references',