A community based topic aggregation platform built on atproto

test(communities): add E2E tests and update service mocks

E2E Tests (3 new test cases):
- Block via XRPC endpoint: Full flow from HTTP → PDS → Jetstream → AppView
- Unblock via XRPC endpoint: Complete unblock flow with DELETE event
- Block fails without authentication: Validates auth requirement (401)

Each E2E test verifies:
✓ XRPC endpoint responds correctly
✓ Record created/deleted on PDS
✓ Jetstream consumer indexes event
✓ AppView database state updated

Unit Test Updates:
- Added 6 mock methods to mockCommunityRepo for blocking operations
- Ensures service layer tests compile and pass

All tests follow existing E2E patterns (subscribe/unsubscribe) for
consistency.

Changed files
+346
tests
+322
tests/integration/community_e2e_test.go
···
t.Logf(" ✓ Subscriber count decremented")
})
+
t.Run("Block via XRPC endpoint", func(t *testing.T) {
+
// Create a community to block
+
community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
+
+
t.Logf("🚫 Blocking community via XRPC endpoint...")
+
blockReq := map[string]interface{}{
+
"community": community.DID,
+
}
+
+
blockJSON, err := json.Marshal(blockReq)
+
if err != nil {
+
t.Fatalf("Failed to marshal block request: %v", err)
+
}
+
+
req, err := http.NewRequest("POST", httpServer.URL+"/xrpc/social.coves.community.blockCommunity", bytes.NewBuffer(blockJSON))
+
if err != nil {
+
t.Fatalf("Failed to create block request: %v", err)
+
}
+
req.Header.Set("Content-Type", "application/json")
+
req.Header.Set("Authorization", "Bearer "+accessToken)
+
+
resp, err := http.DefaultClient.Do(req)
+
if err != nil {
+
t.Fatalf("Failed to POST block: %v", err)
+
}
+
defer func() { _ = resp.Body.Close() }()
+
+
if resp.StatusCode != http.StatusOK {
+
body, readErr := io.ReadAll(resp.Body)
+
if readErr != nil {
+
t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
+
}
+
t.Logf("❌ XRPC Block Failed")
+
t.Logf(" Status: %d", resp.StatusCode)
+
t.Logf(" Response: %s", string(body))
+
t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
+
}
+
+
var blockResp struct {
+
Block struct {
+
RecordURI string `json:"recordUri"`
+
RecordCID string `json:"recordCid"`
+
} `json:"block"`
+
}
+
+
if err := json.NewDecoder(resp.Body).Decode(&blockResp); err != nil {
+
t.Fatalf("Failed to decode block response: %v", err)
+
}
+
+
t.Logf("✅ XRPC block response received:")
+
t.Logf(" RecordURI: %s", blockResp.Block.RecordURI)
+
t.Logf(" RecordCID: %s", blockResp.Block.RecordCID)
+
+
// Extract rkey from URI for verification
+
rkey := ""
+
if uriParts := strings.Split(blockResp.Block.RecordURI, "/"); len(uriParts) >= 4 {
+
rkey = uriParts[len(uriParts)-1]
+
}
+
+
// Verify the block record exists on PDS
+
t.Logf("🔍 Verifying block record exists on PDS...")
+
collection := "social.coves.community.block"
+
pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
+
pdsURL, instanceDID, collection, rkey))
+
if pdsErr != nil {
+
t.Fatalf("Failed to query PDS: %v", pdsErr)
+
}
+
defer func() {
+
if closeErr := pdsResp.Body.Close(); closeErr != nil {
+
t.Logf("Failed to close PDS response: %v", closeErr)
+
}
+
}()
+
+
if pdsResp.StatusCode != http.StatusOK {
+
body, readErr := io.ReadAll(pdsResp.Body)
+
if readErr != nil {
+
t.Fatalf("Block record not found on PDS (status: %d, failed to read body: %v)", pdsResp.StatusCode, readErr)
+
}
+
t.Fatalf("Block record not found on PDS (status: %d): %s", pdsResp.StatusCode, string(body))
+
}
+
t.Logf("✅ Block record exists on PDS")
+
+
// CRITICAL: Simulate Jetstream consumer indexing the block
+
t.Logf("🔄 Simulating Jetstream consumer indexing block event...")
+
blockEvent := jetstream.JetstreamEvent{
+
Did: instanceDID,
+
TimeUS: time.Now().UnixMicro(),
+
Kind: "commit",
+
Commit: &jetstream.CommitEvent{
+
Rev: "test-block-rev",
+
Operation: "create",
+
Collection: "social.coves.community.block",
+
RKey: rkey,
+
CID: blockResp.Block.RecordCID,
+
Record: map[string]interface{}{
+
"subject": community.DID,
+
"createdAt": time.Now().Format(time.RFC3339),
+
},
+
},
+
}
+
if handleErr := consumer.HandleEvent(context.Background(), &blockEvent); handleErr != nil {
+
t.Fatalf("Failed to handle block event: %v", handleErr)
+
}
+
+
// Verify block was indexed in AppView
+
t.Logf("🔍 Verifying block indexed in AppView...")
+
block, err := communityRepo.GetBlock(ctx, instanceDID, community.DID)
+
if err != nil {
+
t.Fatalf("Failed to get block from AppView: %v", err)
+
}
+
if block.RecordURI != blockResp.Block.RecordURI {
+
t.Errorf("RecordURI mismatch: expected %s, got %s", blockResp.Block.RecordURI, block.RecordURI)
+
}
+
+
t.Logf("✅ TRUE E2E BLOCK FLOW COMPLETE:")
+
t.Logf(" Client → XRPC Block → PDS Create → Firehose → Consumer → AppView ✓")
+
t.Logf(" ✓ Block record created on PDS")
+
t.Logf(" ✓ Block indexed in AppView")
+
})
+
+
t.Run("Unblock via XRPC endpoint", func(t *testing.T) {
+
// Create a community and block it first
+
community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
+
+
// Block the community
+
t.Logf("🚫 Blocking community first...")
+
blockReq := map[string]interface{}{
+
"community": community.DID,
+
}
+
blockJSON, err := json.Marshal(blockReq)
+
if err != nil {
+
t.Fatalf("Failed to marshal block request: %v", err)
+
}
+
+
blockHttpReq, err := http.NewRequest("POST", httpServer.URL+"/xrpc/social.coves.community.blockCommunity", bytes.NewBuffer(blockJSON))
+
if err != nil {
+
t.Fatalf("Failed to create block request: %v", err)
+
}
+
blockHttpReq.Header.Set("Content-Type", "application/json")
+
blockHttpReq.Header.Set("Authorization", "Bearer "+accessToken)
+
+
blockResp, err := http.DefaultClient.Do(blockHttpReq)
+
if err != nil {
+
t.Fatalf("Failed to POST block: %v", err)
+
}
+
+
var blockRespData struct {
+
Block struct {
+
RecordURI string `json:"recordUri"`
+
} `json:"block"`
+
}
+
if err := json.NewDecoder(blockResp.Body).Decode(&blockRespData); err != nil {
+
func() { _ = blockResp.Body.Close() }()
+
t.Fatalf("Failed to decode block response: %v", err)
+
}
+
func() { _ = blockResp.Body.Close() }()
+
+
rkey := ""
+
if uriParts := strings.Split(blockRespData.Block.RecordURI, "/"); len(uriParts) >= 4 {
+
rkey = uriParts[len(uriParts)-1]
+
}
+
+
// Index the block via consumer
+
blockEvent := jetstream.JetstreamEvent{
+
Did: instanceDID,
+
TimeUS: time.Now().UnixMicro(),
+
Kind: "commit",
+
Commit: &jetstream.CommitEvent{
+
Rev: "test-block-rev",
+
Operation: "create",
+
Collection: "social.coves.community.block",
+
RKey: rkey,
+
CID: "test-block-cid",
+
Record: map[string]interface{}{
+
"subject": community.DID,
+
"createdAt": time.Now().Format(time.RFC3339),
+
},
+
},
+
}
+
if handleErr := consumer.HandleEvent(context.Background(), &blockEvent); handleErr != nil {
+
t.Fatalf("Failed to handle block event: %v", handleErr)
+
}
+
+
// Now unblock the community
+
t.Logf("✅ Unblocking community via XRPC endpoint...")
+
unblockReq := map[string]interface{}{
+
"community": community.DID,
+
}
+
+
unblockJSON, err := json.Marshal(unblockReq)
+
if err != nil {
+
t.Fatalf("Failed to marshal unblock request: %v", err)
+
}
+
+
req, err := http.NewRequest("POST", httpServer.URL+"/xrpc/social.coves.community.unblockCommunity", bytes.NewBuffer(unblockJSON))
+
if err != nil {
+
t.Fatalf("Failed to create unblock request: %v", err)
+
}
+
req.Header.Set("Content-Type", "application/json")
+
req.Header.Set("Authorization", "Bearer "+accessToken)
+
+
resp, err := http.DefaultClient.Do(req)
+
if err != nil {
+
t.Fatalf("Failed to POST unblock: %v", err)
+
}
+
defer func() { _ = resp.Body.Close() }()
+
+
if resp.StatusCode != http.StatusOK {
+
body, readErr := io.ReadAll(resp.Body)
+
if readErr != nil {
+
t.Fatalf("Expected 200, got %d (failed to read body: %v)", resp.StatusCode, readErr)
+
}
+
t.Logf("❌ XRPC Unblock Failed")
+
t.Logf(" Status: %d", resp.StatusCode)
+
t.Logf(" Response: %s", string(body))
+
t.Fatalf("Expected 200, got %d: %s", resp.StatusCode, string(body))
+
}
+
+
var unblockResp struct {
+
Success bool `json:"success"`
+
}
+
+
if err := json.NewDecoder(resp.Body).Decode(&unblockResp); err != nil {
+
t.Fatalf("Failed to decode unblock response: %v", err)
+
}
+
+
if !unblockResp.Success {
+
t.Errorf("Expected success: true, got: %v", unblockResp.Success)
+
}
+
+
// Verify the block record was deleted from PDS
+
t.Logf("🔍 Verifying block record deleted from PDS...")
+
collection := "social.coves.community.block"
+
pdsResp, pdsErr := http.Get(fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=%s",
+
pdsURL, instanceDID, collection, rkey))
+
if pdsErr != nil {
+
t.Fatalf("Failed to query PDS: %v", pdsErr)
+
}
+
defer func() {
+
if closeErr := pdsResp.Body.Close(); closeErr != nil {
+
t.Logf("Failed to close PDS response: %v", closeErr)
+
}
+
}()
+
+
if pdsResp.StatusCode == http.StatusOK {
+
t.Errorf("❌ Block record still exists on PDS (expected 404, got 200)")
+
} else {
+
t.Logf("✅ Block record successfully deleted from PDS (status: %d)", pdsResp.StatusCode)
+
}
+
+
// CRITICAL: Simulate Jetstream consumer indexing the DELETE event
+
t.Logf("🔄 Simulating Jetstream consumer indexing DELETE event...")
+
deleteEvent := jetstream.JetstreamEvent{
+
Did: instanceDID,
+
TimeUS: time.Now().UnixMicro(),
+
Kind: "commit",
+
Commit: &jetstream.CommitEvent{
+
Rev: "test-unblock-rev",
+
Operation: "delete",
+
Collection: "social.coves.community.block",
+
RKey: rkey,
+
CID: "",
+
Record: nil,
+
},
+
}
+
if handleErr := consumer.HandleEvent(context.Background(), &deleteEvent); handleErr != nil {
+
t.Fatalf("Failed to handle delete event: %v", handleErr)
+
}
+
+
// Verify block was removed from AppView
+
t.Logf("🔍 Verifying block removed from AppView...")
+
_, err = communityRepo.GetBlock(ctx, instanceDID, community.DID)
+
if err == nil {
+
t.Errorf("❌ Block still exists in AppView (should be deleted)")
+
} else if !communities.IsNotFound(err) {
+
t.Fatalf("Unexpected error querying block: %v", err)
+
} else {
+
t.Logf("✅ Block removed from AppView")
+
}
+
+
t.Logf("✅ TRUE E2E UNBLOCK FLOW COMPLETE:")
+
t.Logf(" Client → XRPC Unblock → PDS Delete → Firehose → Consumer → AppView ✓")
+
t.Logf(" ✓ Block deleted from PDS")
+
t.Logf(" ✓ Block removed from AppView")
+
})
+
+
t.Run("Block fails without authentication", func(t *testing.T) {
+
// Create a community to attempt blocking
+
community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
+
+
t.Logf("🔒 Attempting to block community without auth token...")
+
blockReq := map[string]interface{}{
+
"community": community.DID,
+
}
+
+
blockJSON, err := json.Marshal(blockReq)
+
if err != nil {
+
t.Fatalf("Failed to marshal block request: %v", err)
+
}
+
+
req, err := http.NewRequest("POST", httpServer.URL+"/xrpc/social.coves.community.blockCommunity", bytes.NewBuffer(blockJSON))
+
if err != nil {
+
t.Fatalf("Failed to create block request: %v", err)
+
}
+
req.Header.Set("Content-Type", "application/json")
+
// NO Authorization header
+
+
resp, err := http.DefaultClient.Do(req)
+
if err != nil {
+
t.Fatalf("Failed to POST block: %v", err)
+
}
+
defer func() { _ = resp.Body.Close() }()
+
+
// Should fail with 401 Unauthorized
+
if resp.StatusCode != http.StatusUnauthorized {
+
body, _ := io.ReadAll(resp.Body)
+
t.Errorf("Expected 401 Unauthorized, got %d: %s", resp.StatusCode, string(body))
+
} else {
+
t.Logf("✅ Block correctly rejected without authentication (401)")
+
}
+
})
+
t.Run("Update via XRPC endpoint", func(t *testing.T) {
// Create a community first (via service, so it's indexed)
community := createAndIndexCommunity(t, communityService, consumer, instanceDID, pdsURL)
+24
tests/unit/community_service_test.go
···
return nil, nil
}
+
func (m *mockCommunityRepo) BlockCommunity(ctx context.Context, block *communities.CommunityBlock) (*communities.CommunityBlock, error) {
+
return block, nil
+
}
+
+
func (m *mockCommunityRepo) UnblockCommunity(ctx context.Context, userDID, communityDID string) error {
+
return nil
+
}
+
+
func (m *mockCommunityRepo) GetBlock(ctx context.Context, userDID, communityDID string) (*communities.CommunityBlock, error) {
+
return nil, communities.ErrBlockNotFound
+
}
+
+
func (m *mockCommunityRepo) GetBlockByURI(ctx context.Context, recordURI string) (*communities.CommunityBlock, error) {
+
return nil, communities.ErrBlockNotFound
+
}
+
+
func (m *mockCommunityRepo) ListBlockedCommunities(ctx context.Context, userDID string, limit, offset int) ([]*communities.CommunityBlock, error) {
+
return nil, nil
+
}
+
+
func (m *mockCommunityRepo) IsBlocked(ctx context.Context, userDID, communityDID string) (bool, error) {
+
return false, nil
+
}
+
func (m *mockCommunityRepo) CreateMembership(ctx context.Context, membership *communities.Membership) (*communities.Membership, error) {
return membership, nil
}